一级欧美视频_黑巨人与欧美精品一区_精品国产美女_欧洲一区在线电影_清纯唯美日韩_免费在线黄网_波多野结衣一区二区三区_伊人色综合一区二区三区影院视频_一区二区三区日本_av中文在线资源库

本篇博客的內容為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: 8021spring:#給項目來個名字application:name: rabbitmq-provider#配置rabbitMq 服務器rabbitmq:host: 127.0.0.1port: 5672username: rootpassword: root#虛擬host 可以不設置,使用server默認hostvirtual-host: JCcccHost#確認消息已發送到交換機(Exchange)#publisher-confirms: truepublisher-confirm-type: correlated#確認消息已發送到隊列(Queue)publisher-returns: true

完善更多信息

spring:rabbitmq:host: localhostport: 5672virtual-host: /username: guestpassword: guestpublisher-confirm-type: correlatedpublisher-returns: truetemplate:mandatory: trueretry:#發布重試，默認falseenabled: true#重試時間 默認1000msinitial-interval: 1000#重試最大次數 最大3max-attempts: 3#重試最大間隔時間max-interval: 10000#重試的時間隔乘數，比如配2，0第一次等于10s，第二次等于20s，第三次等于40smultiplier: 1listener:\# 默認配置是simpletype: simplesimple:\# 手動ack Acknowledge mode of container. auto noneacknowledge-mode: manual#消費者調用程序線程的最小數量concurrency: 10#消費者最大數量max-concurrency: 10#限制消費者每次只處理一條信息，處理完在繼續下一條prefetch: 1#啟動時是否默認啟動容器auto-startup: true#被拒絕時重新進入隊列default-requeue-rejected: true

@RabbitListener 注解是指定某方法作為消息消費的方法，例如監聽某 Queue里面的消息。

@RabbitListener標注在方法上，直接監聽指定的隊列，此時接收的參數需要與發送市類型一致。

\@Componentpublic 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 {\@RabbitHandlerpublic void processMessage1(String message) {System.out.println(message);}\@RabbitHandlerpublic 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, \@HeadersMap\ headers) {System.out.println(\"body：\"+body);System.out.println(\"Headers：\"+headers);}

\org.springframework.boot\\spring-boot-starter-amqp\\

\@Configurationpublic class RabbitmqConfig {//創建交換機//通過ExchangeBuilder能創建direct、topic、Fanout類型的交換機\@Bean(\"bootExchange\")public Exchange bootExchange() {returnExchangeBuilder.topicExchange(\"zx_topic_exchange\").durable(true).build();}//創建隊列\@Bean(\"bootQueue\")public Queue bootQueue() {return QueueBuilder.durable(\"zx_queue\").build();}/\*\*\* 將隊列與交換機綁定\*\* \@param queue\* \@param exchange\* \@return\*/\@Beanpublic Binding bindQueueExchange(@Qualifier(\"bootQueue\") Queue queue,\@Qualifier(\"bootExchange\") Exchange exchange) {returnBindingBuilder.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 :\*\*/\@Configurationpublic class DirectRabbitConfig {//隊列 起名：TestDirectQueue\@Beanpublic Queue TestDirectQueue() {//durable:是否持久化,默認是false,持久化隊列：會被存儲在磁盤上，當消息代理重啟時仍然存在，暫存隊列：當前連接有效//exclusive:默認也是false，只能被當前創建的連接使用，而且當連接關閉后隊列即被刪除。此參考優先級高于durable//autoDelete:是否自動刪除，當沒有生產者或者消費者使用此隊列，該隊列會自動刪除。// return new Queue(\"TestDirectQueue\",true,true,false);//一般設置一下隊列的持久化就好,其余兩個就是默認falsereturn new Queue(\"TestDirectQueue\",true);}//Direct交換機 起名：TestDirectExchange\@BeanDirectExchange TestDirectExchange() {// return new DirectExchange(\"TestDirectExchange\",true,true);return new DirectExchange(\"TestDirectExchange\",true,false);}//綁定 將隊列和交換機綁定, 并設置用于匹配鍵：TestDirectRouting\@BeanBinding bindingDirect() {returnBindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with(\"TestDirectRouting\");}\@BeanDirectExchange 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 :\*\*/\@Configurationpublic class FanoutRabbitConfig {/\*\*\* 創建三個隊列 ：fanout.A fanout.B fanout.C\* 將三個隊列都綁定在交換機 fanoutExchange 上\* 因為是扇型交換機, 路由鍵無需配置,配置也不起作用\*/\@Beanpublic Queue queueA() {return new Queue(\"fanout.A\");}\@Beanpublic Queue queueB() {return new Queue(\"fanout.B\");}\@Beanpublic Queue queueC() {return new Queue(\"fanout.C\");}\@BeanFanoutExchange fanoutExchange() {return new FanoutExchange(\"fanoutExchange\");}\@BeanBinding bindingExchangeA() {return BindingBuilder.bind(queueA()).to(fanoutExchange());}\@BeanBinding bindingExchangeB() {return BindingBuilder.bind(queueB()).to(fanoutExchange());}\@BeanBinding 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 :\*\*/\@Configurationpublic class TopicRabbitConfig {//綁定鍵public final static String man = \"topic.man\";public final static String woman = \"topic.woman\";\@Beanpublic Queue firstQueue() {return new Queue(TopicRabbitConfig.man);}\@Beanpublic Queue secondQueue() {return new Queue(TopicRabbitConfig.woman);}\@BeanTopicExchange exchange() {return new TopicExchange(\"topicExchange\");}//將firstQueue和topicExchange綁定,而且綁定的鍵值為topic.man//這樣只要是消息攜帶的路由鍵是topic.man,才會分發到該隊列\@BeanBinding bindingExchangeMessage() {return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);}//將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規則topic.#// 這樣只要是消息攜帶的路由鍵是以topic.開頭,都會分發到該隊列\@BeanBinding bindingExchangeMessage2() {returnBindingBuilder.bind(secondQueue()).to(exchange()).with(\"topic.#\");}}

直接發送給隊列

//指定消息隊列的名字，直接發送消息到消息隊列中\@Testpublic void testSimpleQueue() {// 隊列名稱String queueName = \"simple.queue\";// 消息String message = \"hello, spring amqp!\";// 發送消息rabbitTemplate.convertAndSend(queueName, message);}

發送給交換機，然后走不同的模式

////指定交換機的名字，將消息發送給交換機，然后不同模式下，消息隊列根據key得到消息\@Testpublic 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 :\*\*/\@RestControllerpublic class SendMessageController {\@AutowiredRabbitTemplate 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-ddHH:mm:ss\"));Map\ map=new HashMap\<\>();map.put(\"messageId\",messageId);map.put(\"messageData\",messageData);map.put(\"createTime\",createTime);//將消息攜帶綁定鍵值：TestDirectRouting 發送到交換機TestDirectExchangerabbitTemplate.convertAndSend(\"TestDirectExchange\",\"TestDirectRouting\", map);return \"ok\";}}

//使用注解@RabbitListener定義當前方法監聽RabbitMQ中指定名稱的消息隊列。\@Componentpublic class MessageListener {\@RabbitListener(queues = \"direct_queue\")public void receive(String id){System.out.println(\"已完成短信發送業務(rabbitmq direct)，id：\"+id);}}參數用Map接收也可以\@Component\@RabbitListener(queues = \"TestDirectQueue\")//監聽的隊列名稱TestDirectQueuepublic class DirectReceiver {\@RabbitHandlerpublic void process(Map testMessage) {System.out.println(\"DirectReceiver消費者收到消息 : \" +testMessage.toString());}}

有confirm和return兩種

在application.yml中添加以下配置項：

server:port: 8021spring:#給項目來個名字application:name: rabbitmq-provider#配置rabbitMq 服務器rabbitmq:host: 127.0.0.1port: 5672username: rootpassword: root#虛擬host 可以不設置,使用server默認hostvirtual-host: JCcccHost#確認消息已發送到交換機(Exchange)#publisher-confirms: truepublisher-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;\@Configurationpublic class RabbitConfig {\@Beanpublic RabbitTemplate createRabbitTemplate(ConnectionFactoryconnectionFactory){RabbitTemplate rabbitTemplate = new RabbitTemplate();rabbitTemplate.setConnectionFactory(connectionFactory);//設置開啟Mandatory,才能觸發回調函數,無論消息推送結果怎么樣都強制調用回調函數rabbitTemplate.setMandatory(true);rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {\@Overridepublic void confirm(CorrelationData correlationData, boolean ack, Stringcause) {System.out.println(\"ConfirmCallback:\"+\"相關數據：\"+correlationData);System.out.println(\"ConfirmCallback: \"+\"確認情況：\"+ack);System.out.println(\"ConfirmCallback: \"+\"原因：\"+cause);}});rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {\@Overridepublic void returnedMessage(Message message, int replyCode, StringreplyText, 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\@Slf4jpublic class SmsRabbitMqUtils implements RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnsCallback {\@Resourceprivate RedisTemplate\ redisTemplate;\@Resourceprivate RabbitTemplate rabbitTemplate;private String finalId = null;private SmsDTO smsDTO = null;/\*\*\* 發布者確認的回調\*\* \@param correlationData 回調的相關數據。\* \@param b ack為真，nack為假\* \@param s 一個可選的原因，用于nack，如果可用，否則為空。\*/\@Overridepublic void confirm(CorrelationData correlationData, boolean b, Strings) {// 消息發送成功，將redis中消息的狀態（status）修改為1if (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\*/\@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error(\"發生異常，返回消息回調:{}\", returnedMessage);// 發送失敗，放入redis失敗集合中，并刪除集合數據redisTemplate.delete(RedisConstant.SMS_MESSAGE_PREFIX + finalId);redisTemplate.opsForHash().put(RedisConstant.MQ_PRODUCER, finalId,this.smsDTO);}\@PostConstructpublic void init() {rabbitTemplate.setConfirmCallback(this);rabbitTemplate.setReturnsCallback(this);}}

手動確認

yml配置#手動確認 manuallistener:simple:acknowledge-mode: manual

首先在消費者項目中創建MessageListenerConfig

import com.elegant.rabbitmqconsumer.receiver.MyAckReceiver;import org.springframework.amqp.core.AcknowledgeMode;import org.springframework.amqp.core.Queue;importorg.springframework.amqp.rabbit.connection.CachingConnectionFactory;importorg.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;\@Configurationpublic class MessageListenerConfig {\@Autowiredprivate CachingConnectionFactory connectionFactory;\@Autowiredprivate MyAckReceiver myAckReceiver;//消息接收處理類\@Beanpublic SimpleMessageListenerContainer simpleMessageListenerContainer() {SimpleMessageListenerContainer container = newSimpleMessageListenerContainer(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;importorg.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;import org.springframework.stereotype.Component;import java.io.ByteArrayInputStream;import java.io.ObjectInputStream;import java.util.Map;\@Componentpublic class MyAckReceiver implements ChannelAwareMessageListener {\@Overridepublic void onMessage(Message message, Channel channel) throws Exception{long deliveryTag = message.getMessageProperties().getDeliveryTag();try {byte\[\] body = message.getBody();ObjectInputStream ois = new ObjectInputStream(newByteArrayInputStream(body));Map\ msgMap = (Map\) 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;importorg.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;import org.springframework.stereotype.Component;import java.io.ByteArrayInputStream;import java.io.ObjectInputStream;import java.util.Map;\@Componentpublic class MyAckReceiver implements ChannelAwareMessageListener {\@Overridepublic void onMessage(Message message, Channel channel) throws Exception{long deliveryTag = message.getMessageProperties().getDeliveryTag();try {byte\[\] body = message.getBody();ObjectInputStream ois = new ObjectInputStream(newByteArrayInputStream(body));Map\ msgMap = (Map\) 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\@Slf4jpublic class SendSmsListener {\@Resourceprivate RedisTemplate\ redisTemplate;\@Resourceprivate SendSmsUtils sendSmsUtils;/\*\*\* 監聽發送短信普通隊列\* \@param smsDTO\* \@param message\* \@param channel\* \@throws IOException\*/\@RabbitListener(queues = SMS_QUEUE_NAME)public void sendSmsListener(SmsDTO smsDTO, Message message, Channelchannel) 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()) \|\|!\"sendsuccess\".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, Channelchannel) throws IOException {try{log.error(\"監聽到死信隊列消息==\>{}\",sms);channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);}catch (Exception e){channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);}}}

#配置RabbitMQspring:rabbitmq:host: 192.168.126.3port: 5672username: guestpassword: guestvirtual-host: /#開啟自動確認 none 手動確認 manuallistener:simple:#消費端限流機制必須開啟手動確認acknowledge-mode: manual#消費端最多拉取的消息條數，簽收后不滿該條數才會繼續拉取prefetch: 5

可以設置隊列的存活時間，也可以設置具體消息的存活時間

設置隊列中所有消息的存活時間

return QueueBuilder

.durable(QUEUE_NAME)//隊列持久化

.ttl(10000)//設置隊列的所有消息存活10s

.build();

即在創建隊列時，設置存活時間

設置某條消息的存活時間

//發送消息，并設置該消息的存活時間

\@Testpublic void testSendMessage(){//1.創建消息屬性MessageProperties messageProperties = new MessageProperties();//2.設置存活時間messageProperties.setExpiration(\"10000\");//3.創建消息對象Message message = newMessage(\"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) Exchangeexchange, \@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) Exchangeexchange, \@Qualifier(NORMAL_QUEUE) Queue queue){return BindingBuilder.bind(queue).to(exchange).with(\"my_routing\").noargs();}}

RabbitMQ并未實現延遲隊列功能，所以可以通過死信隊列實現延遲隊列的功能

即給普通隊列設置存活時間30分鐘，過期后發送至死信隊列，在死信消費者監聽死信隊列消息，查看訂單狀態，是否支付，未支付則取消訂單，回退庫存即可。

消費者監聽延遲隊列

\@Componentpublic class ExpireOrderConsumer {//監聽過期訂單隊列\@RabbitListener(queues = \"expire_queue\")public void listenMessage(String orderId){//模擬處理數據庫等業務System.out.println(\"查詢\"+orderId+\"號訂單的狀態，如果已支付無需處理，如果未支付則回退庫存\");}}控制層代碼\@RestControllerpublic class OrderController {\@Autowiredprivate 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 \"下單成功，修改庫存\";}}