package org.jeecg.boot.starter.rabbitmq.config;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.jeecg.boot.starter.rabbitmq.core.MapMessageConverter;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.SerializerMessageConverter;
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;
import java.util.HashMap;
import java.util.Map;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 15:55
* @Param
* @Description //TODO rabbitmq配置类
* RabbitMQ四种交换机类型介绍
* 1.Direct Exchange(直连交换机)：处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配
* 2.Fanout Exchange(扇型交换机)：不处理路由键。你只需要简单的将队列绑定到交换机上。一个发送到交换机的消息都会被转发到与该交换机绑定的所有队列上
* 3.Topic Exchange(主题交换机)：将路由键和某模式进行匹配
* 4.Headers Exchanges(头交换机)：不处理路由键。而是根据发送的消息内容中的headers属性进行匹配
*/
@Configuration
public class RabbitConfig {
/*
*如果需要在生产者需要消息发送后的回调，需要对rabbitTemplate设置ConfirmCallback对象，
* 由于不同的生产者需要对应不同的ConfirmCallback，
* 如果rabbitTemplate设置为单例bean，则所有的rabbitTemplate实际的ConfirmCallback为最后一次申明的ConfirmCallback所以这里必须为prototype类型 要不然回报错
*
* singleton作用域：当把一个Bean定义设置为singleton作用域是，Spring IoC容器中只会存在一个共享的Bean实例，并且所有对Bean的请求，
* 只要id与该Bean定义相匹配，则只会返回该Bean的同一实例。值得强调的是singleton作用域是Spring中的缺省作用域。
* prototype作用域：prototype作用域的Bean会导致在每次对该Bean请求（将其注入到另一个Bean中，或者以程序的方式调用容器的getBean()方法）时都会创建一个新的Bean实例。根据经验，对有状态的Bean应使用prototype作用域，而对无状态的Bean则应该使用singleton作用
*/
@Bean
@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
/**
* 当mandatory标志位设置为true时
* 如果exchange根据自身类型和消息routingKey无法找到一个合适的queue存储消息
* 那么broker会调用basic.return方法将消息返还给生产者
* 当mandatory设置为false时，出现上述情况broker会直接将消息丢弃
*/
template.setMandatory(true);
// template.setMessageConverter(new MapMessageConverter());
template.setMessageConverter(new SerializerMessageConverter());
//使用单独的发送连接，避免生产者由于各种原因阻塞而导致消费者同样阻塞
// template.setUsePublisherConnection(true);
/**
* 从2.1版本开始，CorrelationData对象具有ListenableFuture，可用于获取结果，而不是在rabbitTemplate上使用ConfirmCallback,
* ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调
* ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调/消费失败的时候
*/
// template.setConfirmCallback();
// template.setReturnCallback();
return template;
}
}
package org.jeecg.boot.starter.rabbitmq.constant;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 15:56
* @Param
* @Description //TODO rabbitmq常量
* @return
*/
public class MqConstant {
/**
* 测试交换机
*/
public final static String TEST_QUEUE = "test_queue";
public final static String TEST_EXCHANGE = "test_exchange";
public final static String ROUTERKEY = "test_queue";
public final static String TEST_QUEUE1 = "test_queue1";
public final static String ROUTERKEY1 = "test_queue1";
public final static String TEST_EXCHANGE1 = "test_exchange1";
/**
* 直连交互机
*/
public final static String DIRECT_QUEUE = "direct_queue";
public final static String DIRECT_EXCHANGE = "direct_exchange";
public final static String DIRECT_ROUTERKEY = "direct_routerKey";
/**
* 死信队列 出现的情况有三种：1过期的消息，2设置队列里面达到了最大值，新增的消息会进入死信队列，3拒绝的消息，比如：手动ack的时候调用nack并且不重回队列
*/
public final static String DLX_EXCHANGE = "dlx_exchange";
public final static String DLX_QUEUE = "dlx_queue";
public final static String DLX_ROUTERKRY = "dlx_routerkey";
//topic 模式
public final static String TOPIC_EXCHANGE = "topic_exchange";
public final static String TOPIC_QUEUE = "topic_queue";
public final static String TPOIC_ROUTERKEY = "topic.#";
//fanout 模式
public final static String FANOUT_EXCHANGE = "fanout_exchange";
public final static String FANOUT_QUEUE = "fanout_queue";
//延迟队列
public static final Integer delay = 10000;
public static final String LAZY_EXCHANGE = "Lazy_Exchange";
public static final String LAZY_QUEUE = "Lazy_Queue";
public static final String LAZY_KEY = "lazy.#";
}
package org.jeecg.boot.starter.rabbitmq.callback;
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.nio.charset.StandardCharsets;
/**
* @description 自定义消息发送确认的回调
* 实现接口：implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback
* ConfirmCallback：只确认消息是否正确到达交换机中,不管是否到达交换机,该回调都会执行;
* ReturnCallback：如果消息从交换机未正确到达队列中将会执行，正确到达则不执行;
*/
@Component
@Slf4j
public class CustomConfirmAndReturnCallback implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnCallback {
/**
* 消息从交换机成功到达队列，则returnedMessage方法不会执行;
* 消息从交换机未能成功到达队列，则returnedMessage方法会执行;
*/
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
log.info("returnedMessage回调方法>>>" + new String(message.getBody(), StandardCharsets.UTF_8) + ",replyCode:" + replyCode
+ ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);
}
/**
* 如果消息没有到达交换机,则该方法中isSendSuccess = false,error为错误信息;
* 如果消息正确到达交换机,则该方法中isSendSuccess = true;
* correlationData 每个发送的消息都需要配备一个 CorrelationData 相关数据对象，CorrelationData 对象内部只有一个 id 属性，用来表示当前消息唯一性。
*/
@Override
public void confirm(CorrelationData correlationData, boolean isSendSuccess, String error) {
log.info("confirm回调方法>>>回调消息ID为: " + correlationData.getId());
if (isSendSuccess) {
log.info("confirm回调方法>>>消息发送到交换机成功！");
} else {
log.info("confirm回调方法>>>消息发送到交换机失败！，原因 : [{}]", error);
}
}
}
package org.jeecg.boot.starter.rabbitmq.core;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.util.StringUtils;
import java.io.Serializable;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/13 10:42
* @Param
* @Description //TODO 消息队列载体初始化
*/
public class MqMessageFactoryInit {
/**
* 消息载体初始化
*
* @param msg 消息内容
* @param deliveryMode 消息属性:消息投递模式(持久性(默认),将消息写入磁盘或内存队列
* @param headerKey 设置消息的header,类型为Map<String,Object>
* @param headerValue 设置消息的header,类型为Map<String,Object>
* @param expiration 消息的失效时间
* @param delay 延迟时间
* @param contentType 让消费者知道如何解释消息体；
* @param correlationId 消息响应唯一id；
* @param messageId 消息id
* @param priority 消息的优先级（0-9）越大越高
* @param replyTo 消息失败，如果需要重回队列，重回到那个队列
* @return
*/
public static Message init(String msg, MessageDeliveryMode deliveryMode, String headerKey, String headerValue, String expiration, Integer delay, String contentType,
String correlationId, String messageId, Integer priority, String replyTo) {
MessageProperties properties = new MessageProperties();
if (null != deliveryMode) {
properties.setDeliveryMode(deliveryMode);
}
if (!StringUtils.isEmpty(headerKey) && !StringUtils.isEmpty(headerValue)) {
properties.setHeader(headerKey, headerValue);
}
if (!StringUtils.isEmpty(expiration)) {
properties.setExpiration(expiration);
}
if (null != delay) {
properties.setDelay(delay); //设置延迟的时间
}
if (!StringUtils.isEmpty(contentType)) {
properties.setContentType(contentType);
}
if (!StringUtils.isEmpty(correlationId)) {
properties.setCorrelationId(correlationId);
}
if (!StringUtils.isEmpty(messageId)) {
properties.setMessageId(messageId);
}
if (null != priority) {
properties.setPriority(priority);
}
if (!StringUtils.isEmpty(replyTo)) {
properties.setReplyTo(replyTo);
}
return new Message(msg.getBytes(), properties);
}
/**
* 消息响应数据初始化
*
* @param id 响应编号
* @return
*/
public static CorrelationData init(String id) {
return new CorrelationData(id);
}
}
package org.jeecg.boot.starter.rabbitmq.core;
import java.util.HashMap;
import java.util.concurrent.CountDownLatch;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 15:49
* @Param
* @Description //TODO 共享变量操作类，共享变量管理库主要用于解决【指定的】多个线程安全地共享一个变量，而其它线程无法共享，也可以存redis
* @return
*/
public class SharedVariableUtils {
//回调函数结果：Result管理仓库，每个发布都对应一个result，一个主线程，二个回调的子线程共享这个result，执行完后消除
private static HashMap<String, ResultVo> resultMap = new HashMap<>();
//countDownLatch管理仓库，每个发布消息请求有两个回调：confirmCallback和returnCallback，但第二个成功时就不会执行所以很难得到结果，所以这里并没有使用
//每个回调线程和主线程分别共享各自的CountDownLatch，这个主键可以由appId+tenantId+回调类型来区分。
private static HashMap<String, CountDownLatch> countDownLatchMap = new HashMap<>();
//发布失败后执行次数的计数器，需要每个线程单独使用
private static ThreadLocal<Integer> count = ThreadLocal.withInitial(() -> new Integer(0));
public static String confirmCall = "confirm";
public static String returnCall = "return";
//==============================Result=====================================
/**
* 主线程获取到了自己的result结果，删除自己的result，避免内存泄漏导致内存溢出
*
* @param resultName
*/
private static void deleteResult(String resultName) {
resultMap.remove(resultName);
}
/**
* 主线程获取result结果
*
* @param resultName
* @return
*/
public static ResultVo getResult(String resultName) {
ResultVo result = resultMap.get(resultName);
deleteResult(resultName);
return result;
}
/**
* 子线程设置值
*
* @param resultName
* @param key
* @param value
* @return
*/
public static void setResult(String resultName, String key, String value) {
ResultVo result = new ResultVo();
result.put(key, value);
resultMap.put(resultName, result);
}
//========================================countDownLaunch=======================================================
/**
* 主线程进行对自己子线程地专属等待
*
* @param countDownLatchName
*/
public static void await(String countDownLatchName) {
CountDownLatch countDownLatch = countDownLatchMap.get(countDownLatchName);
if (countDownLatch == null) {
countDownLatch = new CountDownLatch(1);
countDownLatchMap.put(countDownLatchName, countDownLatch);
}
try {
countDownLatch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
/**
* 主线程等待结束，删除专属的countDownLatch，对象避免内存泄漏导致内存溢出
*
* @param countDownLatchName
*/
public static void deleteCountDownLatch(String countDownLatchName) {
countDownLatchMap.remove(countDownLatchName);
}
/**
* 子线程调用，告知主线程自己执行完毕
*
* @param countDownLatchName
*/
public static void countDown(String countDownLatchName) {
CountDownLatch countDownLatch = countDownLatchMap.get(countDownLatchName);
if (countDownLatch == null) {
countDownLatch = new CountDownLatch(1);
countDownLatchMap.put(countDownLatchName, countDownLatch);
}
countDownLatch.countDown();
}
// ============================================count值===============================================================
//获取count的值
public static int getCount() {
return count.get();
}
//count++
public static void countUp() {
count.set(count.get() + 1);
}
//设置count
public static void setCount(int i) {
count.set(0);
}
//remove
}
package org.jeecg.boot.starter.rabbitmq.exchange;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 16:35
* @Description //TODO 死性队列交换机
* @return
*/
@Configuration
public class DeadDirectExchangeConfig {
/**
* 创建直连死性队列
*
* @return
*/
@Bean
public Queue directQueue() {
Map<String, Object> agruments = new HashMap<String, Object>();
agruments.put("x-dead-letter-exchange", MqConstant.DLX_EXCHANGE);
// agruments.put("x-message-ttl", 5000);
// queue：queue的名称
// exclusive：排他队列，如果一个队列被声明为排他队列，该队列仅对首次申明它的连接可见，并在连接断开时自动删除。这里需要注意三点：
// 1. 排他队列是基于连接可见的，同一连接的不同信道是可以同时访问同一连接创建的排他队列；
// 2.“首次”，如果一个连接已经声明了一个排他队列，其他连接是不允许建立同名的排他队列的，这个与普通队列不同；
// 3.即使该队列是持久化的，一旦连接关闭或者客户端退出，该排他队列都会被自动删除的，这种队列适用于一个客户端发送读取消息的应用场景。
// autoDelete：自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
return new Queue(MqConstant.DIRECT_QUEUE, true, false, false, agruments);
}
@Bean
public DirectExchange directExchange() {
return new DirectExchange(MqConstant.DIRECT_EXCHANGE, true, false);
}
@Bean
public Binding directBinding() {
return BindingBuilder.bind(directQueue()).to(directExchange()).with(MqConstant.DIRECT_ROUTERKEY);
}
}
package org.jeecg.boot.starter.rabbitmq.exchange;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 16:38
* @Description //TODO 延迟交换机
* @return
*/
@Configuration
public class LazyExchangeConfig {
/**
* 延迟队列
*
* @return
*/
@Bean
public TopicExchange lazyExchange() {
TopicExchange exchange = new TopicExchange(MqConstant.LAZY_EXCHANGE, true, false);
exchange.setDelayed(true); //开启延迟队列
return exchange;
}
@Bean
public Queue lazyQueue() {
return new Queue(MqConstant.LAZY_QUEUE, true);
}
@Bean
public Binding lazyBinding() {
return BindingBuilder.bind(lazyQueue()).to(lazyExchange()).with(MqConstant.LAZY_KEY);
}
}
package org.jeecg.boot.starter.rabbitmq.exchange;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 16:33
* @Description //TODO 测试直连交换机
* @return
*/
@Configuration
public class TestDirectExchangeConfig {
/**
* 创建一个队列
*
* @return
*/
@Bean
public Queue createQueue() {
//durable(耐用的)是为了防止宕机等异常而导致消息无法及时接收设计的。这个对queue无太多影响，但对topic影响比较大。
return new Queue(MqConstant.TEST_QUEUE, true);
}
/**
* 创建直连交换机
* 作用:处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配
*
* @return
*/
@Bean
public DirectExchange createExchange() {
//autoDelete：true自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
return new DirectExchange(MqConstant.TEST_EXCHANGE, true, false);
}
/**
* 绑定关键字，将一个特定的Exchange和一个特定的Queue绑定起来。
*
* @return
*/
@Bean
public Binding binding() {
return BindingBuilder.bind(createQueue()).to(createExchange()).with(MqConstant.ROUTERKEY);
}
/**
* 创建一个队列
*
* @return
*/
@Bean
public Queue createQueue1() {
//durable(耐用的)是为了防止宕机等异常而导致消息无法及时接收设计的。这个对queue无太多影响，但对topic影响比较大。
return new Queue(MqConstant.TEST_QUEUE1, true);
}
/**
* 创建直连交换机
* 作用:处理路由键。需要将一个队列绑定到交换机上，要求该消息与一个特定的路由键完全匹配
*
* @return
*/
@Bean
public DirectExchange createExchange1() {
//autoDelete：true自动删除，如果该队列没有任何订阅的消费者的话，该队列会被自动删除。这种队列适用于临时队列。
return new DirectExchange(MqConstant.TEST_EXCHANGE1, true, false);
}
/**
* 绑定关键字，将一个特定的Exchange和一个特定的Queue绑定起来。
*
* @return
*/
@Bean
public Binding binding1() {
return BindingBuilder.bind(createQueue1()).to(createExchange1()).with(MqConstant.ROUTERKEY1);
}
}
package org.jeecg.boot.starter.rabbitmq.exchange;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 16:37
* @Description //TODO 主题交换机死信队列
* @return
*/
@Configuration
public class TopicExchangeConfig {
@Bean
public Queue dlxQueue() {
return new Queue(MqConstant.DLX_QUEUE, true);
}
/**
* 创建主题交换机
* 将路由键和某模式进行匹配。此时队列需要绑定要一个模式上。符号“#”匹配一个或多个词，符号“”匹配不多不少一个词
*
* @return
*/
@Bean
public TopicExchange dlxExchange() {
return new TopicExchange(MqConstant.DLX_EXCHANGE, true, false);
}
@Bean
public Binding dlxBinding() {
return BindingBuilder.bind(dlxQueue()).to(dlxExchange()).with("#");
}
}
package org.jeecg.modules.pQuery.controller;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.jeecg.modules.pQuery.rabitMq.Order;
import org.jeecg.modules.pQuery.sender.SendMqService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.io.Serializable;
/**
* @version 1.0
* @Author zhaozhiqiang
* @Date 2021/12/10 17:00
* @Param
* @Description //TODO 重写消息队列测试
*/
@Api(tags = "消息队列重写测试")
@RestController
@RequestMapping("/mq")
@CrossOrigin
@Slf4j
public class RabbitMqTest {
@Autowired
private SendMqService sendMqService;
@Value("${spring.rabbitmq.publisher-confirms}")
private String confirms;
/**
* @Description: 测试用的，没有绑定交换机，使用默认的交换机direct模式
*/
@ApiOperation(value = "test1")
@GetMapping(value = "/send")
public void send(String msg) {
log.info("============:"+confirms);
sendMqService.send(msg, MqConstant.ROUTERKEY);
}
/**
* 这个是死信队列，产生死信队列有三种可能，
* 第一：对于一些拒绝消费的而且不重回队列的消息进入死信队列
* 第二：对于一些过期的队列
* 第三：对于已经设置队列的最大消息值的，再次进入队列的
* 绑定死信队列的要点就是在声明队列的时候设置参数 x-dead-letter-exchange 指向对应的交换机就可以了
*
* @param msg
*/
@ApiOperation(value = "死信队列")
@GetMapping(value = "/dlsSend")
public void dlsSend(String msg) {
sendMqService.dlsSend(msg, MqConstant.DIRECT_ROUTERKEY);
}
/**
* 一对多的模式，设置routingkey为xx.#或者xx.* #号表示可以包含多个参数 * 表示可以包含一个参数
*
* @param msg
*/
@ApiOperation(value = "一对多的模式")
@GetMapping(value = "/sendTopic")
public void sendTopic(String msg) {
sendMqService.sendTopic(msg, "topic.message");
}
/**
* 广播模式：不用设置还是不设置routingkey 都可以消费
*/
@ApiOperation(value = "广播模式")
@GetMapping(value = "/fanoutSender")
public void fanoutSender() {
Order order = new Order(2,"你好，我是fanout队列");
sendMqService.fanoutSender(order);
}
/**
* 延迟队列，使用的rabbitmq的一个插件：rabbitmq_delayed_message_exchange 需要下载下来放到plugins文件夹里面
* 然后启动这个插件，然后在创建queue的时候开启延迟队列就可以达到延迟效果，本人亲测正常使用
* @param msg
*/
@ApiOperation(value = "延迟队列")
@GetMapping(value = "/lazySender")
public void lazySender(String msg) {
sendMqService.lazySender(msg,MqConstant.LAZY_KEY);
}
/**
* 测试死性队列/事务回滚/重试机制
* 正常队列与死信交换机绑定给队列设置参数：x-dead-letter-exchange和x-dead-letter-routing-key
*/
@ApiOperation(value = "测试死性队列/事务回滚/重试机制")
@GetMapping(value = "/testDb")
public void testDb() {
sendMqService.testDb();
}
}
package org.jeecg.modules.pQuery.receive;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.jeecg.boot.starter.rabbitmq.listenter.MqListener;
import org.jeecg.boot.starter.rabbitmq.receive.BaseReceive;
import org.jeecg.common.base.BaseMap;
import org.jeecg.common.config.mqtoken.UserTokenContext;
import org.jeecg.common.websocket.DebugContext;
import org.jeecg.modules.pQuery.controller.RabbitMqTest;
import org.jeecg.modules.pQuery.mapper.PeOrderInfoMapper;
import org.jeecg.modules.pQuery.rabitMq.Order;
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.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import java.io.IOException;
import java.util.Map;
/**
* rabbitmq消费者
* Channel api参数
* 1 channel.exchangeDeclare(String exchange, String type, boolean durable, boolean autoDelete,Map<String, Object> arguments)：
* type：有direct、fanout、topic三种
* durable：true、false true：服务器重启会保留下来Exchange。警告：仅设置此选项，不代表消息持久化。即不保证重启后消息还在。
* autoDelete:true、false.true:当已经没有消费者时，服务器是否可以删除该Exchange。
* 2 channel.basicQos(int prefetchSize, int prefetchCount, boolean global)
* prefetchSize：0
* prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
* global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别
* 3 channel.basicPublish(String exchange, String routingKey, boolean mandatory, boolean immediate, BasicProperties props, byte[] body)
* routingKey：路由键，#匹配0个或多个单词，*匹配一个单词，在topic exchange做消息转发用
* mandatory：true：如果exchange根据自身类型和消息routeKey无法找到一个符合条件的queue，那么会调用basic.return方法将消息返还给生产者。false：出现上述情形broker会直接将消息扔掉
* 4 channel.basicAck(long deliveryTag, boolean multiple)
* deliveryTag:该消息的index
* multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息。
* 5 channel.basicNack(delivery.getEnvelope().getDeliveryTag(), false, true);
* deliveryTag:该消息的index
* multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。
* requeue：被拒绝的是否重新入队列
* 6 channel.basicReject(delivery.getEnvelope().getDeliveryTag(), false);
* deliveryTag:该消息的index
* requeue：被拒绝的是否重新入队列
* channel.basicNack 与 channel.basicReject 的区别在于basicNack可以拒绝多条消息，而basicReject一次只能拒绝一条消息
* 7 channel.basicConsume(QUEUE_NAME, true, consumer);
* autoAck：是否自动ack，如果不自动ack，需要使用channel.ack、channel.nack、channel.basicReject 进行消息应答
*/
@Component
@Slf4j
public class ReceiveMqService {
@Autowired
PeOrderInfoMapper peOrderInfoMapper;
private String token = UserTokenContext.getToken();
@RabbitListener(queues = MqConstant.TEST_QUEUE)
public void customer(Message message, Channel channel) throws IOException {
UserTokenContext.setToken(token);
long deliveryTag = message.getMessageProperties().getDeliveryTag();
/**
* prefetchSize：0
* prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
* global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别
* prefetchSize 和global这两项，rabbitmq没有实现，暂且不研究
*/
channel.basicQos(0, 1, false);
log.info("deliveryTag:{}", deliveryTag);
Integer num = (Integer) message.getMessageProperties().getHeaders().get("num");
if (num == 0) {
//不回队列一个个不消费
channel.basicNack(deliveryTag, false, false);
} else {
log.info("basicAck:{}", "一个个消费");
//一个个消费
channel.basicAck(deliveryTag, false);
}
}
@RabbitListener(queues = MqConstant.DIRECT_QUEUE)
public void customer2(Message message, Channel channel) throws IOException, InterruptedException {
UserTokenContext.setToken(token);
channel.basicQos(0, 1, false);
String correlationId = message.getMessageProperties().getCorrelationId();
System.out.println(correlationId + "*******************");
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.err.println("-----------consume message----------");
System.err.println("body: " + new String(message.getBody()));
log.info("deliveryTag为：" + deliveryTag);
// channel.basicAck(deliveryTag, false);
channel.basicReject(deliveryTag, false);
}
@RabbitListener(queues = MqConstant.DLX_QUEUE)
public void dlxCustomer(Message message, Channel channel) throws IOException, InterruptedException {
UserTokenContext.setToken(token);
channel.basicQos(0, 1, false);
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.err.println("-----------consume message----------");
System.err.println("body: " + new String(message.getBody()));
Integer num = (Integer) message.getMessageProperties().getHeaders().get("num");
log.info("死信队列里面的deliveryTag为：" + deliveryTag);
log.info("私信队列准备ack消息了");
channel.basicAck(deliveryTag, false);
}
@RabbitListener(
bindings = @QueueBinding(
exchange = @Exchange(value = MqConstant.TOPIC_EXCHANGE, type = ExchangeTypes.TOPIC,
durable = "true", autoDelete = "false"),
value = @Queue(value = MqConstant.TOPIC_QUEUE, durable = "true"),
key = MqConstant.TPOIC_ROUTERKEY))
public void topicCustomer(Message message, Channel channel) throws IOException {
UserTokenContext.setToken(token);
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.out.println(deliveryTag + "*************");
channel.basicAck(deliveryTag, false);
}
/**
* @param order payload是一种以JSON格式进行数据传输的一种方式。
* @param map delivery_tag是消息投递序号，每个channel对应一个(long类型)，从1开始到9223372036854775807范围，在手动消息确认时可以对指定delivery_tag的消息进行ack、nack、reject等操作
* @param channel
* @throws IOException
*/
@RabbitListener(bindings = @QueueBinding(
exchange = @Exchange(value = MqConstant.FANOUT_EXCHANGE, type = ExchangeTypes.FANOUT, durable = "true"),
value = @Queue(value = MqConstant.FANOUT_QUEUE, durable = "true")))
public void fanoutCustomer(@Payload Order order, @Headers Map<String, Object> map, Channel channel,Message message) throws IOException {
UserTokenContext.setToken(token);
System.out.println(order.getId());
Long delivery = (Long) map.get(AmqpHeaders.DELIVERY_TAG);
log.info("批次： {},序列号:{}", delivery,message.getMessageProperties().getDeliveryTag());
try {
System.out.println("消费成功");
channel.basicAck(delivery, false);
} catch (IOException e) {
System.out.println("消费失败");
channel.basicNack(
delivery, false, false
);
}
}
@RabbitListener(queues = MqConstant.LAZY_QUEUE)
public void lazyCustomer(Message message, Channel channel) {
UserTokenContext.setToken(token);
long deliveryTag = message.getMessageProperties().getDeliveryTag();
System.err.println("body: " + new String(message.getBody()));
try {
channel.basicAck(deliveryTag, false);
} catch (IOException e) {
try {
channel.basicNack(deliveryTag, false, false);
} catch (IOException e1) {
log.info("失败");
}
}
}
/**
* mq消息队列重试机制
* 1 必须抛出异常才能重试
* 2 消息不能重新进入队列，所有参数设置false
* 重试的意义:
* 1场景需要
* 2防止服务器内存溢出
* @param msg
* @param channel
* @param message
* @throws IOException
*/
@RabbitListener(queues = MqConstant.TEST_QUEUE1)
@Transactional
public void customer1(String msg, Channel channel, Message message) throws IOException {
UserTokenContext.setToken(token);
/**
* prefetchSize：0
* prefetchCount：会告诉RabbitMQ不要同时给一个消费者推送多于N个消息，即一旦有N个消息还没有ack，则该consumer将block掉，直到有消息ack
* global：true\false 是否将上面设置应用于channel，简单点说，就是上面限制是channel级别的还是consumer级别
* prefetchSize 和global这两项，rabbitmq没有实现，暂且不研究
*/
try {
// 这里模拟一个空指针异常，
log.info("【Consumer01】批次： {}", message.getMessageProperties().getDeliveryTag());
log.info("【Consumer01成功接收到消息】>>> {}", msg);
peOrderInfoMapper.setValues();
String string = null;
string.length();
// 确认收到消息，只确认当前消费者的一个消息收到
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
} catch (Exception e) {
e.printStackTrace();
//次重复消息投递
log.info("【Consumer01】批次： {}", message.getMessageProperties().getDeliveryTag());
// 拒绝消息，并且不再重新进入队列,重试机制两个条件1抛出异常，2不能重入队列而是重试，此处设置false
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
throw e;
}
}
}
package org.jeecg.modules.pQuery.sender;
import lombok.extern.slf4j.Slf4j;
import org.jeecg.boot.starter.rabbitmq.callback.CustomConfirmAndReturnCallback;
import org.jeecg.boot.starter.rabbitmq.constant.MqConstant;
import org.jeecg.common.base.BaseMap;
import org.jeecg.modules.pQuery.controller.RabbitMqTest;
import org.jeecg.modules.pQuery.rabitMq.Order;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
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 javax.annotation.PostConstruct;
import java.util.UUID;
/**
* rabbitmq生成者
*/
@Component
@Slf4j
public class SendMqService extends CustomConfirmAndReturnCallback {
@Autowired
private RabbitTemplate rabbitTemplate;
// @Autowired
// public SendMqService(RabbitTemplate amqpTemplate) {
// amqpTemplate.setConfirmCallback(this::confirm);
// amqpTemplate.setReturnCallback(this::returnedMessage);
// this.rabbitTemplate = amqpTemplate;
// }
/**
* PostConstruct: 用于在依赖关系注入完成之后需要执行的方法上，以执行任何初始化.
*/
@PostConstruct
public void init() {
//指定 ConfirmCallback
rabbitTemplate.setConfirmCallback(this);
//指定 ReturnCallback
rabbitTemplate.setReturnCallback(this);
}
/**
* 消费者消费确认后回调
* @param correlationData 每个发送的消息都需要配备一个 CorrelationData 相关数据对象，CorrelationData 对象内部只有一个 id 属性，用来表示当前消息唯一性。
* @param ack 是否确认消费
* @param cause 失败原因
*/
public void confirm(CorrelationData correlationData, boolean ack, String cause) {
super.confirm(correlationData, ack, cause);
log.info("数据编号========correlationData:{}======ack:{}", ack, correlationData);
if (!ack) {
log.info("异常处理...."+cause);
}
}
/**
* 交换机和路由键匹配不了后return回调
* 如果消息从交换机未正确到达队列中将会执行，正确到达则不执行
* @param message
* @param replyCode
* @param replyText
* @param exchange
* @param routingKey
*/
public void returnedMessage(Message message, int replyCode,
String replyText, String exchange, String routingKey) {
super.returnedMessage(message, replyCode, replyText, exchange, routingKey);
log.info("return exchange: {}, routingKey: {}, replyCode: {}, replyText: {}",
exchange, routingKey, replyCode, replyText);
}
public void send(String msg, String routingKey) {
for (int i = 0; i <= 5; i++) {
BaseMap map = new BaseMap();
map.put("num", i);
// MessageProperties properties = new MessageProperties();
// properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
// properties.setHeader("num", i);
// Message message = new Message(msg.getBytes(), map);
// amqpTemplate.convertAndSend(MqConstant.TEST_EXCHANGE, routingKey, map);
// amqpTemplate.convertAndSend(MqConstant.TEST_EXCHANGE, routingKey, map, message -> {
// return message;
// });
MessageProperties properties = new MessageProperties();
properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
properties.setHeader("num",i);
Message message = new Message(msg.getBytes(), properties);
rabbitTemplate.convertAndSend(MqConstant.TEST_EXCHANGE, routingKey, message);
}
}
public void dlsSend(String msg, String directRouterkey) {
for (int i = 0; i <= 5; i++) {
MessageProperties properties = new MessageProperties();
properties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
properties.setHeader("num",i);
properties.setExpiration("5000");
Message message = new Message(msg.getBytes(), properties);
String a = UUID.randomUUID().toString();
CorrelationData correlationData = new CorrelationData(String.valueOf(i));
rabbitTemplate.convertAndSend(MqConstant.DIRECT_EXCHANGE, MqConstant.DIRECT_ROUTERKEY, message,correlationData);
}
}
public void sendTopic(String msg, String routingKey) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
rabbitTemplate.convertAndSend(MqConstant.TOPIC_EXCHANGE,routingKey,msg,correlationData);
}
public void fanoutSender(Order order) {
CorrelationData correlationData = new CorrelationData(order.getId().toString());
rabbitTemplate.convertAndSend(MqConstant.FANOUT_EXCHANGE,null,order,correlationData);
}
public void lazySender(String msg, String routingKey) {
MessageProperties messageProperties = new MessageProperties();
messageProperties.setDelay(MqConstant.delay); //设置延迟的时间
//设置消息投递模式持久性和临时性，临时性重启会丢失或者没有接收者会丢失
messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
Message message = new Message(msg.getBytes(),messageProperties);
rabbitTemplate.convertAndSend(MqConstant.LAZY_EXCHANGE,routingKey,message);
}
public void testDb() {
for (int i = 0; i < 1; i++) {
CorrelationData correlationData = new CorrelationData(UUID.randomUUID().toString());
// log.info("【Producer】发送的消费ID = {}", correlationData.getId());
String msg = "hello confirm message" + i;
// log.info("【Producer】发送的消息 = {}", msg);
rabbitTemplate.convertAndSend(MqConstant.TEST_EXCHANGE1, MqConstant.ROUTERKEY1, msg, correlationData);
}
}
}
依赖
<!-- 消息总线 rabbitmq -->
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-starter-bus-amqp</artifactId>
</dependency>
配置
#rabbitmq配置
rabbitmq:
host: pe-boot-rabbitmq
username: admin
password: admin
port: 5672
virtual-host: /
connection-timeout: 15000
#开启confirm模式
publisher-confirms: true
#开启return模式，前提是下面的mandatory设置为true否则会删除消息
publisher-returns: true
#消费者端开启自动ack模式
template.mandatory: true
#新版本publisher-confirms已经修改为publisher-confirm-type，默认为NONE,CORRELATED值是发布消息成功到交换器会触发回调
publisher-confirm-type: correlated
listener:
simple:
acknowledge-mode: manual
#并发消费者的最大值
max-concurrency: 5
#并发消费者的初始化值
concurrency: 1
#每个消费者每次监听时可拉取
prefetch: 1
# 重试机制
retry:
#是否开启消费者重试
enabled: true
#最大重试次数
max-attempts: 5
#重试间隔时间（单位毫秒）
initial-interval: 5000
#重试最大时间间隔（单位毫秒）
max-interval: 1200000
#间隔时间乘子，间隔时间*乘子=下一次的间隔时间，最大不能超过设置的最大间隔时间
multiplier: 2
原文链接：https://blog.csdn.net/weixin_38501485/article/details/121910347
发表评论