本篇博客的内容为RabbitMQ在开发过程中的快速上手使用,侧重于代码部分,几乎没有相关概念的介绍,相关概念请参考以下csdn博客,两篇都是我找的精华帖,供大家学习。本篇博客也持续更新~~~
内容代码部分由于word转md格式有些问题,可以直接查看我的有道云笔记,链接:https://note.youdao.com/s/Ab7Cjiu
csdn博客:
基础部分:https://blog.csdn.net/qq_35387940/article/details/100514134
高级部分:https://blog.csdn.net/weixin_49076273/article/details/124991012
server: port: 8021 spring: #给项目来个名字 application: name: rabbitmq-provider #配置rabbitMq 服务器 rabbitmq: host: 127.0.0.1 port: 5672 username: root password: root #虚拟host 可以不设置,使用server默认host virtual-host: JCcccHost #确认消息已发送到交换机(Exchange) #publisher-confirms: true publisher-confirm-type: correlated #确认消息已发送到队列(Queue) publisher-returns: true
完善更多信息
spring: rabbitmq: host: localhost port: 5672 virtual-host: / username: guest password: guest publisher-confirm-type: correlated publisher-returns: true template: mandatory: true retry: #发布重试,默认false enabled: true #重试时间 默认1000ms initial-interval: 1000 #重试最大次数 最大3 max-attempts: 3 #重试最大间隔时间 max-interval: 10000 #重试的时间隔乘数,比如配2,0 第一次等于10s,第二次等于20s,第三次等于40s multiplier: 1 listener: \# 默认配置是simple type: simple simple: \# 手动ack Acknowledge mode of container. auto none acknowledge-mode: manual #消费者调用程序线程的最小数量 concurrency: 10 #消费者最大数量 max-concurrency: 10 #限制消费者每次只处理一条信息,处理完在继续下一条 prefetch: 1 #启动时是否默认启动容器 auto-startup: true #被拒绝时重新进入队列 default-requeue-rejected: true
@RabbitListener 注解是指定某方法作为消息消费的方法,例如监听某 Queue
里面的消息。
@RabbitListener标注在方法上,直接监听指定的队列,此时接收的参数需要与发送市类型一致。
\@Component public class PointConsumer { //监听的队列名 \@RabbitListener(queues = \"point.to.point\") public void processOne(String name) { System.out.println(\"point.to.point:\" + name); } }
@RabbitListener 可以标注在类上面,需配合 @RabbitHandler 注解一起使用
@RabbitListener 标注在类上面表示当有收到消息的时候,就交给
@RabbitHandler 的方法处理,根据接受的参数类型进入具体的方法中。
\@Component \@RabbitListener(queues = \"consumer_queue\") public class Receiver { \@RabbitHandler public void processMessage1(String message) { System.out.println(message); } \@RabbitHandler public void processMessage2(byte\[\] message) { System.out.println(new String(message)); } }
@Payload
可以获取消息中的 body 信息
\@RabbitListener(queues = \"debug\") public void processMessage1(@Payload String body) { System.out.println(\"body:\"+body); }
@Header,@Headers
可以获得消息中的 headers 信息
\@RabbitListener(queues = \"debug\") public void processMessage1(@Payload String body, \@Header String token) { System.out.println(\"body:\"+body); System.out.println(\"token:\"+token); } \@RabbitListener(queues = \"debug\") public void processMessage1(@Payload String body, \@Headers Map\<String,Object\> headers) { System.out.println(\"body:\"+body); System.out.println(\"Headers:\"+headers); }
<dependency\> \<groupId\>org.springframework.boot\</groupId\> \<artifactId\>spring-boot-starter-amqp\</artifactId\> \</dependency\>
\@Configuration public class RabbitmqConfig { //创建交换机 //通过ExchangeBuilder能创建direct、topic、Fanout类型的交换机 \@Bean(\"bootExchange\") public Exchange bootExchange() { return ExchangeBuilder.topicExchange(\"zx_topic_exchange\").durable(true).build(); } //创建队列 \@Bean(\"bootQueue\") public Queue bootQueue() { return QueueBuilder.durable(\"zx_queue\").build(); } /\*\* \* 将队列与交换机绑定 \* \* \@param queue \* \@param exchange \* \@return \*/ \@Bean public Binding bindQueueExchange(@Qualifier(\"bootQueue\") Queue queue, \@Qualifier(\"bootExchange\") Exchange exchange) { return BindingBuilder.bind(queue).to(exchange).with(\"boot.#\").noargs(); } }
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; /\*\* \* \@Author : JCccc \* \@CreateTime : 2019/9/3 \* \@Description : \*\*/ \@Configuration public class DirectRabbitConfig { //队列 起名:TestDirectQueue \@Bean public Queue TestDirectQueue() { // durable:是否持久化,默认是false,持久化队列:会被存储在磁盘上,当消息代理重启时仍然存在,暂存队列:当前连接有效 // exclusive:默认也是false,只能被当前创建的连接使用,而且当连接关闭后队列即被删除。此参考优先级高于durable // autoDelete:是否自动删除,当没有生产者或者消费者使用此队列,该队列会自动删除。 // return new Queue(\"TestDirectQueue\",true,true,false); //一般设置一下队列的持久化就好,其余两个就是默认false return new Queue(\"TestDirectQueue\",true); } //Direct交换机 起名:TestDirectExchange \@Bean DirectExchange TestDirectExchange() { // return new DirectExchange(\"TestDirectExchange\",true,true); return new DirectExchange(\"TestDirectExchange\",true,false); } //绑定 将队列和交换机绑定, 并设置用于匹配键:TestDirectRouting \@Bean Binding bindingDirect() { return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(\"TestDirectRouting\"); } \@Bean DirectExchange lonelyDirectExchange() { return new DirectExchange(\"lonelyDirectExchange\"); } }
import org.springframework.amqp.core.Binding; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /\*\* \* \@Author : JCccc \* \@CreateTime : 2019/9/3 \* \@Description : \*\*/ \@Configuration public class FanoutRabbitConfig { /\*\* \* 创建三个队列 :fanout.A fanout.B fanout.C \* 将三个队列都绑定在交换机 fanoutExchange 上 \* 因为是扇型交换机, 路由键无需配置,配置也不起作用 \*/ \@Bean public Queue queueA() { return new Queue(\"fanout.A\"); } \@Bean public Queue queueB() { return new Queue(\"fanout.B\"); } \@Bean public Queue queueC() { return new Queue(\"fanout.C\"); } \@Bean FanoutExchange fanoutExchange() { return new FanoutExchange(\"fanoutExchange\"); } \@Bean Binding bindingExchangeA() { return BindingBuilder.bind(queueA()).to(fanoutExchange()); } \@Bean Binding bindingExchangeB() { return BindingBuilder.bind(queueB()).to(fanoutExchange()); } \@Bean Binding bindingExchangeC() { return BindingBuilder.bind(queueC()).to(fanoutExchange()); } }
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; /\*\* \* \@Author : JCccc \* \@CreateTime : 2019/9/3 \* \@Description : \*\*/ \@Configuration public class TopicRabbitConfig { //绑定键 public final static String man = \"topic.man\"; public final static String woman = \"topic.woman\"; \@Bean public Queue firstQueue() { return new Queue(TopicRabbitConfig.man); } \@Bean public Queue secondQueue() { return new Queue(TopicRabbitConfig.woman); } \@Bean TopicExchange exchange() { return new TopicExchange(\"topicExchange\"); } //将firstQueue和topicExchange绑定,而且绑定的键值为topic.man //这样只要是消息携带的路由键是topic.man,才会分发到该队列 \@Bean Binding bindingExchangeMessage() { return BindingBuilder.bind(firstQueue()).to(exchange()).with(man); } //将secondQueue和topicExchange绑定,而且绑定的键值为用上通配路由键规则topic.# // 这样只要是消息携带的路由键是以topic.开头,都会分发到该队列 \@Bean Binding bindingExchangeMessage2() { return BindingBuilder.bind(secondQueue()).to(exchange()).with(\"topic.#\"); } }
直接发送给队列
//指定消息队列的名字,直接发送消息到消息队列中 \@Test public void testSimpleQueue() { // 队列名称 String queueName = \"simple.queue\"; // 消息 String message = \"hello, spring amqp!\"; // 发送消息 rabbitTemplate.convertAndSend(queueName, message); }
发送给交换机,然后走不同的模式
////指定交换机的名字,将消息发送给交换机,然后不同模式下,消息队列根据key得到消息 \@Test public void testSendDirectExchange() { // 交换机名称,有三种类型 String exchangeName = \"itcast.direct\"; // 消息 String message = \"红色警报!日本乱排核废水,导致海洋生物变异,惊现哥斯拉!\"; // 发送消息,red为队列的key,因此此队列会得到消息 rabbitTemplate.convertAndSend(exchangeName, \"red\", message); }
也可以将发送的消息封装到HashMap中然后发送给交换机
import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; import java.util.HashMap; import java.util.Map; import java.util.UUID; /\*\* \* \@Author : JCccc \* \@CreateTime : 2019/9/3 \* \@Description : \*\*/ \@RestController public class SendMessageController { \@Autowired RabbitTemplate rabbitTemplate; //使用RabbitTemplate,这提供了接收/发送等等方法 \@GetMapping(\"/sendDirectMessage\") public String sendDirectMessage() { String messageId = String.valueOf(UUID.randomUUID()); String messageData = \"test message, hello!\"; String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern(\"yyyy-MM-dd HH:mm:ss\")); Map\<String,Object\> map=new HashMap\<\>(); map.put(\"messageId\",messageId); map.put(\"messageData\",messageData); map.put(\"createTime\",createTime); //将消息携带绑定键值:TestDirectRouting 发送到交换机TestDirectExchange rabbitTemplate.convertAndSend(\"TestDirectExchange\", \"TestDirectRouting\", map); return \"ok\"; } }
//使用注解@RabbitListener定义当前方法监听RabbitMQ中指定名称的消息队列。 \@Component public class MessageListener { \@RabbitListener(queues = \"direct_queue\") public void receive(String id){ System.out.println(\"已完成短信发送业务(rabbitmq direct),id:\"+id); } } 参数用Map接收也可以 \@Component \@RabbitListener(queues = \"TestDirectQueue\")//监听的队列名称 TestDirectQueue public class DirectReceiver { \@RabbitHandler public void process(Map testMessage) { System.out.println(\"DirectReceiver消费者收到消息 : \" + testMessage.toString()); } }
有confirm和return两种
在application.yml中添加以下配置项:
server: port: 8021 spring: #给项目来个名字 application: name: rabbitmq-provider #配置rabbitMq 服务器 rabbitmq: host: 127.0.0.1 port: 5672 username: root password: root #虚拟host 可以不设置,使用server默认host virtual-host: JCcccHost #确认消息已发送到交换机(Exchange) #publisher-confirms: true publisher-confirm-type: correlated #确认消息已发送到队列(Queue) publisher-returns: true
有两种配置方法:
写到配置类中
写到工具类或者普通类中,但是这个类得实现那两个接口
编写消息确认回调函数
import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.connection.CorrelationData; import org.springframework.amqp.rabbit.core.RabbitTemplate; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; \@Configuration public class RabbitConfig { \@Bean public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){ RabbitTemplate rabbitTemplate = new RabbitTemplate(); rabbitTemplate.setConnectionFactory(connectionFactory); //设置开启Mandatory,才能触发回调函数,无论消息推送结果怎么样都强制调用回调函数 rabbitTemplate.setMandatory(true); rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { \@Override public void confirm(CorrelationData correlationData, boolean ack, String cause) { System.out.println(\"ConfirmCallback: \"+\"相关数据:\"+correlationData); System.out.println(\"ConfirmCallback: \"+\"确认情况:\"+ack); System.out.println(\"ConfirmCallback: \"+\"原因:\"+cause); } }); rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() { \@Override public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) { System.out.println(\"ReturnCallback: \"+\"消息:\"+message); System.out.println(\"ReturnCallback: \"+\"回应码:\"+replyCode); System.out.println(\"ReturnCallback: \"+\"回应信息:\"+replyText); System.out.println(\"ReturnCallback: \"+\"交换机:\"+exchange); System.out.println(\"ReturnCallback: \"+\"路由键:\"+routingKey); } }); return rabbitTemplate; } }
\@Component \@Slf4j public class SmsRabbitMqUtils implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback { \@Resource private RedisTemplate\<String, String\> redisTemplate; \@Resource private RabbitTemplate rabbitTemplate; private String finalId = null; private SmsDTO smsDTO = null; /\*\* \* 发布者确认的回调 \* \* \@param correlationData 回调的相关数据。 \* \@param b ack为真,nack为假 \* \@param s 一个可选的原因,用于nack,如果可用,否则为空。 \*/ \@Override public void confirm(CorrelationData correlationData, boolean b, String s) { // 消息发送成功,将redis中消息的状态(status)修改为1 if (b) { redisTemplate.opsForHash().put(RedisConstant.SMS_MESSAGE_PREFIX + finalId, \"status\", 1); } else { // 发送失败,放入redis失败集合中,并删除集合数据 log.error(\"短信消息投送失败:{}\--\>{}\", correlationData, s); redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + finalId); redisTemplate.opsForHash().put(RedisConstant.MQ_PRODUCER, finalId, this.smsDTO); } } /\*\* \* 发生异常时的消息返回提醒 \* \* \@param returnedMessage \*/ \@Override public void returnedMessage(ReturnedMessage returnedMessage) { log.error(\"发生异常,返回消息回调:{}\", returnedMessage); // 发送失败,放入redis失败集合中,并删除集合数据 redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + finalId); redisTemplate.opsForHash().put(RedisConstant.MQ_PRODUCER, finalId, this.smsDTO); } \@PostConstruct public void init() { rabbitTemplate.setConfirmCallback(this); rabbitTemplate.setReturnsCallback(this); } }
手动确认
yml配置 #手动确认 manual listener: simple: acknowledge-mode: manual
首先在消费者项目中创建MessageListenerConfig
import com.elegant.rabbitmqconsumer.receiver.MyAckReceiver; import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.Queue; import org.springframework.amqp.rabbit.connection.CachingConnectionFactory; import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; \@Configuration public class MessageListenerConfig { \@Autowired private CachingConnectionFactory connectionFactory; \@Autowired private MyAckReceiver myAckReceiver;//消息接收处理类 \@Bean public SimpleMessageListenerContainer simpleMessageListenerContainer() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); container.setConcurrentConsumers(1); container.setMaxConcurrentConsumers(1); container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默认是自动确认,这里改为手动确认消息 //设置一个队列 container.setQueueNames(\"TestDirectQueue\"); //如果同时设置多个如下: 前提是队列都是必须已经创建存在的 // container.setQueueNames(\"TestDirectQueue\",\"TestDirectQueue2\",\"TestDirectQueue3\"); //另一种设置队列的方法,如果使用这种情况,那么要设置多个,就使用addQueues //container.setQueues(new Queue(\"TestDirectQueue\",true)); //container.addQueues(new Queue(\"TestDirectQueue2\",true)); //container.addQueues(new Queue(\"TestDirectQueue3\",true)); container.setMessageListener(myAckReceiver); return container; } }
然后创建手动确认监听类MyAckReceiver(手动确认模式需要实现ChannelAwareMessageListener)
import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; import org.springframework.stereotype.Component; import java.io.ByteArrayInputStream; import java.io.ObjectInputStream; import java.util.Map; \@Component public class MyAckReceiver implements ChannelAwareMessageListener { \@Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { byte\[\] body = message.getBody(); ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(body)); Map\<String,String\> msgMap = (Map\<String,String\>) ois.readObject(); String messageId = msgMap.get(\"messageId\"); String messageData = msgMap.get(\"messageData\"); String createTime = msgMap.get(\"createTime\"); ois.close(); System.out.println(\" MyAckReceiver messageId:\"+messageId+\" messageData:\"+messageData+\" createTime:\"+createTime); System.out.println(\"消费的主题消息来自:\"+message.getMessageProperties().getConsumerQueue()); channel.basicAck(deliveryTag, true); //第二个参数,手动确认可以被批处理,当该参数为 true 时,则可以一次性确认 delivery_tag 小于等于传入值的所有消息 //channel.basicReject(deliveryTag, true);//第二个参数,true会重新放回队列,所以需要自己根据业务逻辑判断什么时候使用拒绝 } catch (Exception e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } } }
如果想实现不同的队列,有不同的监听确认处理机制,做不同的业务处理,那么这样做:
首先需要在配置类中绑定队列,然后只需要根据消息来自不同的队列名进行区分处理即可
import com.rabbitmq.client.Channel; import org.springframework.amqp.core.Message; import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener; import org.springframework.stereotype.Component; import java.io.ByteArrayInputStream; import java.io.ObjectInputStream; import java.util.Map; \@Component public class MyAckReceiver implements ChannelAwareMessageListener { \@Override public void onMessage(Message message, Channel channel) throws Exception { long deliveryTag = message.getMessageProperties().getDeliveryTag(); try { byte\[\] body = message.getBody(); ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(body)); Map\<String,String\> msgMap = (Map\<String,String\>) ois.readObject(); String messageId = msgMap.get(\"messageId\"); String messageData = msgMap.get(\"messageData\"); String createTime = msgMap.get(\"createTime\"); ois.close(); if (\"TestDirectQueue\".equals(message.getMessageProperties().getConsumerQueue())){ System.out.println(\"消费的消息来自的队列名为:\"+message.getMessageProperties().getConsumerQueue()); System.out.println(\"消息成功消费到 messageId:\"+messageId+\" messageData:\"+messageData+\" createTime:\"+createTime); System.out.println(\"执行TestDirectQueue中的消息的业务处理流程\...\...\"); } if (\"fanout.A\".equals(message.getMessageProperties().getConsumerQueue())){ System.out.println(\"消费的消息来自的队列名为:\"+message.getMessageProperties().getConsumerQueue()); System.out.println(\"消息成功消费到 messageId:\"+messageId+\" messageData:\"+messageData+\" createTime:\"+createTime); System.out.println(\"执行fanout.A中的消息的业务处理流程\...\...\"); } channel.basicAck(deliveryTag, true); //channel.basicReject(deliveryTag, true);//为true会重新放回队列 } catch (Exception e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } } }
\@Component \@Slf4j public class SendSmsListener { \@Resource private RedisTemplate\<String, String\> redisTemplate; \@Resource private SendSmsUtils sendSmsUtils; /\*\* \* 监听发送短信普通队列 \* \@param smsDTO \* \@param message \* \@param channel \* \@throws IOException \*/ \@RabbitListener(queues = SMS_QUEUE_NAME) public void sendSmsListener(SmsDTO smsDTO, Message message, Channel channel) throws IOException { String messageId = message.getMessageProperties().getMessageId(); int retryCount = (int) redisTemplate.opsForHash().get(RedisConstant.SMS_MESSAGE_PREFIX + messageId, \"retryCount\"); if (retryCount \> 3) { //重试次数大于3,直接放到死信队列 log.error(\"短信消息重试超过3次:{}\", messageId); //basicReject方法拒绝deliveryTag对应的消息,第二个参数是否requeue,true则重新入队列,否则丢弃或者进入死信队列。 //该方法reject后,该消费者还是会消费到该条被reject的消息。 channel.basicReject(message.getMessageProperties().getDeliveryTag(),false); redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + messageId); return; } try { String phoneNum = smsDTO.getPhoneNum(); String code = smsDTO.getCode(); if(StringUtils.isAnyBlank(phoneNum,code)){ throw new RuntimeException(\"sendSmsListener参数为空\"); } // 发送消息 SendSmsResponse sendSmsResponse = sendSmsUtils.sendSmsResponse(phoneNum, code); SendStatus\[\] sendStatusSet = sendSmsResponse.getSendStatusSet(); SendStatus sendStatus = sendStatusSet\[0\]; if(!\"Ok\".equals(sendStatus.getCode()) \|\|!\"send success\".equals(sendStatus.getMessage())){ throw new RuntimeException(\"发送验证码失败\"); } //手动确认消息 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); log.info(\"短信发送成功:{}\",smsDTO); redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + messageId); } catch (Exception e) { redisTemplate.opsForHash().put(RedisConstant.SMS_MESSAGE_PREFIX+messageId,\"retryCount\",retryCount+1); channel.basicReject(message.getMessageProperties().getDeliveryTag(),true); } } /\*\* \* 监听到发送短信死信队列 \* \@param sms \* \@param message \* \@param channel \* \@throws IOException \*/ \@RabbitListener(queues = SMS_DELAY_QUEUE_NAME) public void smsDelayQueueListener(SmsDTO sms, Message message, Channel channel) throws IOException { try{ log.error(\"监听到死信队列消息==\>{}\",sms); channel.basicAck(message.getMessageProperties().getDeliveryTag(),false); }catch (Exception e){ channel.basicReject(message.getMessageProperties().getDeliveryTag(),true); } } }
#配置RabbitMQ spring: rabbitmq: host: 192.168.126.3 port: 5672 username: guest password: guest virtual-host: / #开启自动确认 none 手动确认 manual listener: simple: #消费端限流机制必须开启手动确认 acknowledge-mode: manual #消费端最多拉取的消息条数,签收后不满该条数才会继续拉取 prefetch: 5
可以设置队列的存活时间,也可以设置具体消息的存活时间
设置队列中所有消息的存活时间
return QueueBuilder
.durable(QUEUE_NAME)//队列持久化
.ttl(10000)//设置队列的所有消息存活10s
.build();
即在创建队列时,设置存活时间
设置某条消息的存活时间
//发送消息,并设置该消息的存活时间
\@Test public void testSendMessage() { //1.创建消息属性 MessageProperties messageProperties = new MessageProperties(); //2.设置存活时间 messageProperties.setExpiration(\"10000\"); //3.创建消息对象 Message message = new Message(\"sendMessage\...\".getBytes(),messageProperties); //4.发送消息 rabbitTemplate.convertAndSend(\"my_topic_exchange1\",\"my_routing\",message); }
若设置中间的消息的存活时间,当过期时,该消息不会被移除,但是该消息已经不会被消费了,需要等到该消息到队里顶端才会被移除。因为队列是头出,尾进,故而要移除它需要等到它在顶端时才可以。
在队列设置存活时间,也在单条消息设置存活时间,则以时间短的为准
死信队列和普通队列没有任何区别,只需要将普通队列需要绑定死信交换机和死信队列就能够实现功能
import org.springframework.amqp.core.\*; import org.springframework.beans.factory.annotation.Qualifier; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; \@Configuration//Rabbit配置类 public class RabbitConfig4 { private final String DEAD_EXCHANGE = \"dead_exchange\"; private final String DEAD_QUEUE = \"dead_queue\"; private final String NORMAL_EXCHANGE = \"normal_exchange\"; private final String NORMAL_QUEUE = \"normal_queue\"; //创建死信交换机 \@Bean(DEAD_EXCHANGE) public Exchange deadExchange() { return ExchangeBuilder .topicExchange(DEAD_EXCHANGE)//交换机类型 ;参数为名字 topic为通配符模式的交换机 .durable(true)//是否持久化,true即存到磁盘,false只在内存上 .build(); } //创建死信队列 \@Bean(DEAD_QUEUE) public Queue deadQueue() { return QueueBuilder .durable(DEAD_QUEUE)//队列持久化 //.maxPriority(10)//设置队列的最大优先级,最大可以设置255,但官网推荐不超过10,太高比较浪费资源 .build(); } //死信交换机绑定死信队列 \@Bean //@Qualifier注解,使用名称装配进行使用 public Binding bindDeadQueue(@Qualifier(DEAD_EXCHANGE) Exchange exchange, \@Qualifier(DEAD_QUEUE) Queue queue) { return BindingBuilder .bind(queue) .to(exchange) .with(\"dead_routing\") .noargs(); } //创建普通交换机 \@Bean(NORMAL_EXCHANGE) public Exchange normalExchange() { return ExchangeBuilder .topicExchange(NORMAL_EXCHANGE)//交换机类型 ;参数为名字 topic为通配符模式的交换机 .durable(true)//是否持久化,true即存到磁盘,false只在内存上 .build(); } //创建普通队列 \@Bean(NORMAL_QUEUE) public Queue normalQueue() { return QueueBuilder .durable(NORMAL_QUEUE)//队列持久化 //.maxPriority(10)//设置队列的最大优先级,最大可以设置255,但官网推荐不超过10,太高比较浪费资源 .deadLetterExchange(DEAD_EXCHANGE)//绑定死信交换机 .deadLetterRoutingKey(\"dead_routing\")//死信队列路由关键字 .ttl(10000)//消息存活10s .maxLength(10)//队列最大长度为10 .build(); } //普通交换机绑定普通队列 \@Bean //@Qualifier注解,使用名称装配进行使用 public Binding bindNormalQueue(@Qualifier(NORMAL_EXCHANGE) Exchange exchange, \@Qualifier(NORMAL_QUEUE) Queue queue) { return BindingBuilder .bind(queue) .to(exchange) .with(\"my_routing\") .noargs(); } }
RabbitMQ并未实现延迟队列功能,所以可以通过死信队列实现延迟队列的功能
即给普通队列设置存活时间30分钟,过期后发送至死信队列,在死信消费者监听死信队列消息,查看订单状态,是否支付,未支付则取消订单,回退库存即可。
消费者监听延迟队列
\@Component public class ExpireOrderConsumer { //监听过期订单队列 \@RabbitListener(queues = \"expire_queue\") public void listenMessage(String orderId) { //模拟处理数据库等业务 System.out.println(\"查询\"+orderId+\"号订单的状态,如果已支付无需处理,如果未支付则回退库存\"); } } 控制层代码 \@RestController public class OrderController { \@Autowired private RabbitTemplate rabbitTemplate; \@RequestMapping(value = \"/place/{orderId}\",method = RequestMethod.GET) public String placeOrder(@PathVariable String orderId) { //模拟service层处理 System.out.println(\"处理订单数据\...\"); //将订单id发送到订单队列 rabbitTemplate.convertAndSend(\"order_exchange\",\"order_routing\",orderId); return \"下单成功,修改库存\"; } }