springboot2.5.6集成RabbitMq实现Topic主题模式(推荐)
作者:Scarlet-Max 发布时间:2021-10-03 22:01:22
标签:springboot,RabbitMq,Topic,主题模式
1.application.yml
server:
port: 8184
spring:
application:
name: rabbitmq-demo
rabbitmq:
host: 127.0.0.1 # ip地址
port: 5672
username: admin # 连接账号
password: 123456 # 连接密码
template:
retry:
enabled: true # 开启失败重试
initial-interval: 10000ms # 第一次重试的间隔时长
max-interval: 300000ms # 最长重试间隔,超过这个间隔将不再重试
multiplier: 2 # 下次重试间隔的倍数,此处是2即下次重试间隔是上次的2倍
exchange: topic.exchange # 缺省的交换机名称,此处配置后,发送消息如果不指定交换机就会使用这个
publisher-confirm-type: correlated # 生产者确认机制,确保消息会正确发送,如果发送失败会有错误回执,从而触发重试
publisher-returns: true
listener:
type: simple
simple:
acknowledge-mode: manual
prefetch: 1 # 限制每次发送一条数据。
concurrency: 3 # 同一个队列启动几个消费者
max-concurrency: 3 # 启动消费者最大数量
# 重试策略相关配置
retry:
enabled: true # 是否支持重试
max-attempts: 5
stateless: false
multiplier: 1.0 # 时间策略乘数因子
initial-interval: 1000ms
max-interval: 10000ms
default-requeue-rejected: true
2.pom.xml引入依赖
<!-- rabbitmq -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
3.常量类创建
/**
* @author kkp
* @ClassName RabbitMqConstants
* @date 2021/11/3 14:16
* @Description
*/
public class RabbitMqConstants {
public final static String TEST1_QUEUE = "test1-queue";
public final static String TEST2_QUEUE = "test2-queue";
public final static String EXCHANGE_NAME = "test.topic.exchange";
/**
* routingKey1
*/
public final static String TOPIC_TEST1_ROUTINGKEY = "topic.test1.*";
public final static String TOPIC_TEST1_ROUTINGKEY_TEST = "topic.test1.test";
/**
* routingKey1
*/
public final static String TOPIC_TEST2_ROUTINGKEY = "topic.test2.*";
public final static String TOPIC_TEST2_ROUTINGKEY_TEST = "topic.test2.test";
}
4.配置Configuration
import com.example.demo.common.RabbitMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
/**
* @author kkp
* @ClassName RabbitMqConfig
* @date 2021/11/3 14:16
* @Description
*/
@Slf4j
@Configuration
public class RabbitMqConfig {
@Autowired
private CachingConnectionFactory connectionFactory;
/**
* 声明交换机
*/
@Bean(RabbitMqConstants.EXCHANGE_NAME)
public Exchange exchange(){
//durable(true) 持久化,mq重启之后交换机还在
// Topic模式
//return ExchangeBuilder.topicExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build();
//发布订阅模式
return ExchangeBuilder.fanoutExchange(RabbitMqConstants.EXCHANGE_NAME).durable(true).build();
}
/**
* 声明队列
* new Queue(QUEUE_EMAIL,true,false,false)
* durable="true" 持久化 rabbitmq重启的时候不需要创建新的队列
* auto-delete 表示消息队列没有在使用时将被自动删除 默认是false
* exclusive 表示该消息队列是否只在当前connection生效,默认是false
*/
@Bean(RabbitMqConstants.TEST1_QUEUE)
public Queue esQueue() {
return new Queue(RabbitMqConstants.TEST1_QUEUE);
}
/**
* 声明队列
*/
@Bean(RabbitMqConstants.TEST2_QUEUE)
public Queue gitalkQueue() {
return new Queue(RabbitMqConstants.TEST2_QUEUE);
}
/**
* TEST1_QUEUE队列绑定交换机,指定routingKey
*/
@Bean
public Binding bindingEs(@Qualifier(RabbitMqConstants.TEST1_QUEUE) Queue queue,
@Qualifier(RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY).noargs();
}
/**
* TEST2_QUEUE队列绑定交换机,指定routingKey
*/
@Bean
public Binding bindingGitalk(@Qualifier(RabbitMqConstants.TEST2_QUEUE) Queue queue,
@Qualifier(RabbitMqConstants.EXCHANGE_NAME) Exchange exchange) {
return BindingBuilder.bind(queue).to(exchange).with(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY).noargs();
}
/**
* 如果需要在生产者需要消息发送后的回调,
* 需要对rabbitTemplate设置ConfirmCallback对象,
* 由于不同的生产者需要对应不同的ConfirmCallback,
* 如果rabbitTemplate设置为单例bean,
* 则所有的rabbitTemplate实际的ConfirmCallback为最后一次申明的ConfirmCallback。
* @return
*/
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplate() {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
return template;
}
}
5.Rabbit工具类创建
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import java.util.UUID;
/**
* @author kkp
* @ClassName RabbitMqUtils
* @date 2021/11/3 14:21
* @Description
*/
@Slf4j
@Component
public class RabbitMqUtils implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback{
private RabbitTemplate rabbitTemplate;
/**
* 构造方法注入
*/
@Autowired
public RabbitMqUtils(RabbitTemplate rabbitTemplate) {
this.rabbitTemplate = rabbitTemplate;
//这是是设置回调能收到发送到响应
rabbitTemplate.setConfirmCallback(this);
//如果设置备份队列则不起作用
rabbitTemplate.setMandatory(true);
rabbitTemplate.setReturnCallback(this);
}
/**
* 回调确认
*/
@Override
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
if(ack){
log.info("消息发送成功:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
}else{
log.info("消息发送失败:correlationData({}),ack({}),cause({})",correlationData,ack,cause);
}
}
/**
* 消息发送到转换器的时候没有对列,配置了备份对列该回调则不生效
* @param message
* @param replyCode
* @param replyText
* @param exchange
* @param routingKey
*/
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}",exchange,routingKey,replyCode,replyText,message);
}
/**
* 发送到指定Queue
* @param queueName
* @param obj
*/
public void send(String queueName, Object obj){
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
this.rabbitTemplate.convertAndSend(queueName, obj, correlationId);
}
/**
* 1、交换机名称
* 2、routingKey
* 3、消息内容
*/
public void sendByRoutingKey(String exChange, String routingKey, Object obj){
CorrelationData correlationId = new CorrelationData(UUID.randomUUID().toString());
this.rabbitTemplate.convertAndSend(exChange, routingKey, obj, correlationId);
}
}
6.service创建
public interface TestService {
String sendTest1(String content);
String sendTest2(String content);
}
7.impl实现
import com.example.demo.common.RabbitMqConstants;
import com.example.demo.util.RabbitMqUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
/**
* @author kkp
* @ClassName TestServiceImpl
* @date 2021/11/3 14:24
* @Description
*/
@Service
@Slf4j
public class TestServiceImpl implements TestService {
@Autowired
private RabbitMqUtils rabbitMqUtils;
@Override
public String sendTest1(String content) {
rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME,
RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST, content);
log.info(RabbitMqConstants.TOPIC_TEST1_ROUTINGKEY_TEST+"***************发送成功*****************");
return "发送成功!";
}
@Override
public String sendTest2(String content) {
rabbitMqUtils.sendByRoutingKey(RabbitMqConstants.EXCHANGE_NAME,
RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST, content);
log.info(RabbitMqConstants.TOPIC_TEST2_ROUTINGKEY_TEST+"***************发送成功*****************");
return "发送成功!";
}
}
8.监听类
import com.example.demo.common.RabbitMqConstants;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ExchangeTypes;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;
/**
* @author kkp
* @ClassName RabbitMqListener
* @date 2021/11/3 14:22
* @Description
*/
@Slf4j
@Component
public class RabbitMqListener {
@RabbitListener(queues = RabbitMqConstants.TEST1_QUEUE)
public void test1Consumer(Message message, Channel channel) {
try {
//手动确认消息已经被消费
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
log.info("Counsoum1消费消息:" + message.toString() + "。成功!");
} catch (Exception e) {
e.printStackTrace();
log.info("Counsoum1消费消息:" + message.toString() + "。失败!");
}
}
@RabbitListener(queues = RabbitMqConstants.TEST2_QUEUE)
public void test2Consumer(Message message, Channel channel) {
try {
//手动确认消息已经被消费
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
log.info("Counsoum2消费消息:" + message.toString() + "。成功!");
} catch (Exception e) {
e.printStackTrace();
log.info("Counsoum2消费消息:" + message.toString() + "。失败!");
}
}
}
9.Controller测试
import com.example.demo.server.TestService;
import jdk.nashorn.internal.objects.annotations.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import java.util.Map;
/**
* @author kkp
* @ClassName TestController
* @date 2021/11/3 14:25
* @Description
*/
@Slf4j
@RestController
@RequestMapping("/enterprise")
public class TestController {
@Autowired
private TestService testService;
@GetMapping("/finance")
public String hello3(@RequestParam(required = false) Map<String, Object> params) {
return testService.sendTest2(params.get("entId").toString());
}
/**
* 发送消息test2
* @param content
* @return
*/
@PostMapping(value = "/finance2")
public String sendTest2(@RequestBody String content) {
return testService.sendTest2(content);
}
}
来源:https://blog.csdn.net/weixin_44907173/article/details/121124048
0
投稿
猜你喜欢
- Spring中有个RestTemplate类用来发送HTTP请求很方便,本文分享一个SpringBoot发送POST请求并接收返回数据的例子
- 前言随着标准Java的版本更新,开发者总是可以从升级后的版本中获取想要的功能。本文将给大家详细介绍下mac下面的java9版本安装使用,分享
- 简单回顾一下CAS算法CAS算法 即compare and swap(比较与交换),是一种有名的无锁算法。无锁编程,即不使用锁的情况下实现多
- 本文实例为大家分享了C#异步调用的具体代码,供大家参考,具体内容如下using System;using System.Collection
- Get请求1.简单发送Get请求/// <summary>/// 指定Url地址使用Get 方式获取全部字符串/// </
- 异常日志[com.alibaba.dubbo.rpc.filter.TimeoutFilter] - [DUBBO] invok
- 背景前段时间同事碰到一个问题,需要在 SpringCloud 的 Feign 调用中使用自定义的 URL;通常情况下是没有这个需求的;毕竟都
- 一,准备沙箱环境1,登录支付宝,进入 应用列表界面 https://openhome.alipay.com/dev/workspace2,如
- Java选择的泛型类型叫做类型擦除式泛型。什么是类型擦除式泛型呢?就是Java语言中的泛型只存在于程序源码之中,在编译后的字节码文件里,则全
- 一、setting.xml文件的位置今天我们来谈谈Maven setting文件配置的禅定之道。不知道大家有没有听说过禅宗?嗯,没错,就是那
- Spring AOP对嵌套方法不起作用今天在调研系统操作记录日志时,好多教程都是借助于Spring AOP机制来实现。于是也采用这种方法来实
- 目录 1.ReentrantLock可重入锁概述2.可重入3.可打断4.锁超时5.公平锁6.条件变量 Condition1.Reentran
- 一、稀疏数组1、什么是稀疏数组当一个数组中大部分元素为0,或者为同一个值的数组时,可以用稀疏数组来保存该数组。稀疏数组,记录一共有几行几列,
- 目录无SpringMVC全局异常时的流程图SpringMVC全局异常流程图其实是一个ModelAndView对象配置文件applicatio
- 目录引言API介绍1、Optional(),empty(),of(),ofNullable()2、orElse(),orElseGet()和
- 这篇文章主要介绍了Java GZIP压缩与解压缩代码实例,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的
- 本文实例讲述了Java基于解释器模式实现定义一种简单的语言功能。分享给大家供大家参考,具体如下:一 模式定 * 释器模式:就是给定一个语言的文
- 前言Webp是Google推出的一种新型图片格式,相比于 传统的PNG/JPG图片有着更小体积的优势,在Web中有着广泛的应用。由于Webp
- 在基于Mybatis-plus实现多租户架构中,介绍了在多租户项目中如果要开启一个子线程,那么需要手动进行RequestAttributes
- 在为什么阿里巴巴不建议在for循环中使用”+”进行字符串拼接一文中,我们介绍了几种Java中字符串拼接的方式,以及优缺点。其中还有一个重要的