-c 轉發 包裝 let err 接下來 rec ger str

本篇主要記錄Spring boot 集成Rabbitmq，分為兩部分， 第一部分為創建普通消息隊列， 第二部分為延時消息隊列實現：

spring boot提供對mq消息隊列支持amqp相關包，引入即可：

[html] view plain copy

1. <!-- rabbit mq -->
2. <dependency>
3. <groupId>org.springframework.boot</groupId>
4. <artifactId>spring-boot-starter-amqp</artifactId>
5. </dependency>

屬性配置文件application.properties:

[plain] view plain copy

1. #rabbitmq
2. spring.rabbitmq.host=127.0.0.1
3. spring.rabbitmq.port=5672
4. spring.rabbitmq.username=root
5. spring.rabbitmq.password=root

RabbitMq配置類,配置連接工廠以及操作對象:

[java] view plain copy

1. @Configuration
2. @ConfigurationProperties(prefix = "spring.rabbitmq")
3. public class RabbitMQConfiguration {
5. private static Logger logger = Logger.getLogger(RabbitMQConfiguration.class);
7. private String host;
9. private int port;
11. private String username;
13. private String password;
15. // 鏈接信息
16. @Bean
17. public ConnectionFactory connectionFactory() {
18. CachingConnectionFactory connectionFactory = new CachingConnectionFactory(host, port);
19. connectionFactory.setUsername(username);
20. connectionFactory.setPassword(password);
21. connectionFactory.setVirtualHost("/");
22. connectionFactory.setPublisherConfirms(true);
23. logger.info("Create ConnectionFactory bean ..");
24. return connectionFactory;
25. }
27. @Bean
28. @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
29. public RabbitTemplate rabbitTemplate() {
30. RabbitTemplate template = new RabbitTemplate(connectionFactory());
31. return template;
32. }
33. //省略getter setter

[java] view plain copy

1. }

定義Service接口如下：

暫時不考慮延時隊列，定義發送消息接口

[java] view plain copy

1. /**
2. *
3. * @author victor
4. * @desc 消息隊列服務接口
5. */
6. public interface IMessageQueueService {
8. /**
9. * 發送消息到隊列
10. * @param queue 隊列名稱
11. * @param message 消息內容
12. */
13. public void send(String queueName,String message);
16. }

Service實現

[java] view plain copy

1. package com.ks.server.service.impl.queue;
3. import org.springframework.amqp.AmqpException;
4. import org.springframework.amqp.core.Message;
5. import org.springframework.amqp.core.MessagePostProcessor;
6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.stereotype.Service;
10. import com.base.common.codec.JSONUtils;
11. import com.ks.common.constant.MQConstant;
12. import com.ks.common.service.queue.IMessageQueueService;
13. import com.ks.modal.queue.DLXMessage;
15. /**
16. *
17. * @author victor
18. * @desc 消息隊列服務接口實現
19. */
20. @Service("messageQueueService")
21. public class MessageQueueServiceImpl implements IMessageQueueService{
23. @Autowired
24. private RabbitTemplate rabbitTemplate;
26. @Override
27. public void send(String queueName, String msg) {
28. rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,queueName, msg);
29. }
33. }

相關常量類:

[java] view plain copy

1. package com.ks.common.constant;
3. /**
4. *
5. * @author victor
6. * @desc Rabbit消息隊列相關常量
7. */
8. public final class MQConstant {
10. private MQConstant(){
11. }
13. //exchange name
14. public static final String DEFAULT_EXCHANGE = "KSHOP";
16. //DLX QUEUE
17. public static final String DEFAULT_DEAD_LETTER_QUEUE_NAME = "kshop.dead.letter.queue";
19. //DLX repeat QUEUE 死信轉發隊列
20. public static final String DEFAULT_REPEAT_TRADE_QUEUE_NAME = "kshop.repeat.trade.queue";
23. //Hello 測試消息隊列名稱
24. public static final String HELLO_QUEUE_NAME = "HELLO";
27. }

到現在為止,隊列相關配置，以及使用以及封裝完成，接下來是創建隊列，

這裏我是單獨創建一個配置類，用於隊列配置， 創建Hello隊列示例如下:

[java] view plain copy

1. package com.ks.ons.config;
3. import java.util.HashMap;
4. import java.util.Map;
6. import org.springframework.amqp.core.Binding;
7. import org.springframework.amqp.core.BindingBuilder;
8. import org.springframework.amqp.core.DirectExchange;
9. import org.springframework.amqp.core.Queue;
10. import org.springframework.context.annotation.Bean;
11. import org.springframework.context.annotation.Configuration;
13. import com.ks.common.constant.MQConstant;
15. /**
16. *
17. * @author victor
18. * @desc 隊列配置
19. */
20. @Configuration
21. public class QueueConfiguration {
23. //信道配置
24. @Bean
25. public DirectExchange defaultExchange() {
26. return new DirectExchange(MQConstant.DEFAULT_EXCHANGE, true, false);
27. }
30. /********************* hello 隊列 測試 *****************/
31. @Bean
32. public Queue queue() {
33. Queue queue = new Queue(MQConstant.HELLO_QUEUE_NAME,true);
34. return queue;
35. }
37. @Bean
38. public Binding binding() {
39. return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.HELLO_QUEUE_NAME);
40. }
42. }

通過配置隊列bean，在程序啟動時會在rabbitmq中創建相關隊列，啟動程序，可以在rabbtmq管理界面看到信道和隊列信息:

眾所周知，既然有了隊列，用來處理業務的最終還是需要消費者，消費者創建示例如下:

[java] view plain copy

1. package com.ks.ons.processor.hello;
3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
5. import org.springframework.stereotype.Component;
7. import com.ks.common.constant.MQConstant;
9. /**
10. *
11. * @author victor
12. * @desc hello 消息隊列消費者
13. */
14. @Component
15. @RabbitListener(queues = MQConstant.HELLO_QUEUE_NAME)
16. public class HelloProcessor {
18. @RabbitHandler
19. public void process(String content) {
20. System.out.println("接受消息:" + content);
21. }
22. }

註入service

[java] view plain copy

1. @Autowired
2. private IMessageQueueService messageQueueService;

發送消息

[java] view plain copy

1. messageQueueService.send(MQConstant.HELLO_QUEUE_NAME, "測試發送消息");

接下來展示如何實現延時隊列，在此之前如果讀者像我一樣對rabbitmq隊列了解程度並不深入的話，--> 推薦文章， 可以對rabbitmq延時隊列實現思路有大概了解.

在本文中，主要是通過rabbitmq的DLX特性來實現發送延時隊列：

思路如下：

客戶端：指具體往MQ發生消息端， 客戶端將消息內容進行自定義包裝， 將消息中附帶目標隊列名稱。如：客戶端向隊列Q1發送字符串“hello” , 延時時間為60秒， 包裝後修改為{"queueName":"Q1","body": “hello”}，此時，將消息發送到DLX死信隊列，而非Q1隊列，並將消息設置為60秒超時。

DLX：死信隊列，用來存儲有超時時間信息的消息， 並且可以設置當消息超時時，轉發到另一個指定隊列(此處設置轉發到router), 無消費者，當接收到客戶端消息之後，等待消息超時，將消息轉發到指定的Router隊列

Router: 轉發隊列，用來接收死信隊列超時消息， 如上示例消息，在接收到之後，消費者將消息解析，獲取queueName，body,再向所獲取的queueName隊列發送一條消息，內容為body.

Q1,Q2,Q3.： 用戶業務隊列，當Q1收到hello，已經是60秒之後，再進行消費

修改上面代碼 ， 新增兩個隊列，

死信隊列：存放發送的延時消息，

路由轉發隊列：用於接受死信消息死亡， 並將消息轉發到業務目標隊列

修改之後代碼如下：

[java] view plain copy

1. /**
2. *
3. * @author victor
4. * @desc 隊列配置
5. */
6. @Configuration
7. public class QueueConfiguration {
9. //信道配置
10. @Bean
11. public DirectExchange defaultExchange() {
12. return new DirectExchange(MQConstant.DEFAULT_EXCHANGE, true, false);
13. }
16. @Bean
17. public Queue repeatTradeQueue() {
18. Queue queue = new Queue(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME,true,false,false);
19. return queue;
20. }
22. @Bean
23. public Binding drepeatTradeBinding() {
24. return BindingBuilder.bind(repeatTradeQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME);
25. }
27. @Bean
28. public Queue deadLetterQueue() {
29. Map<String, Object> arguments = new HashMap<>();
30. arguments.put("x-dead-letter-exchange", MQConstant.DEFAULT_EXCHANGE);
31. arguments.put("x-dead-letter-routing-key", MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME);
32. Queue queue = new Queue(MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME,true,false,false,arguments);
33. System.out.println("arguments :" + queue.getArguments());
34. return queue;
35. }
37. @Bean
38. public Binding deadLetterBinding() {
39. return BindingBuilder.bind(deadLetterQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME);
40. }
44. /********************* hello 隊列 測試 *****************/
45. @Bean
46. public Queue queue() {
47. Queue queue = new Queue(MQConstant.HELLO_QUEUE_NAME,true);
48. return queue;
49. }
51. @Bean
52. public Binding binding() {
53. return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.HELLO_QUEUE_NAME);
54. }
56. }

修改Service服務：

[java] view plain copy

1. package com.ks.common.service.queue;
4. /**
5. *
6. * @author victor
7. * @desc 消息隊列服務接口
8. */
9. public interface IMessageQueueService {
11. /**
12. * 發送消息到隊列
13. * @param queue 隊列名稱
14. * @param message 消息內容
15. */
16. public void send(String queueName,String message);
18. /**
19. * 延遲發送消息到隊列
20. * @param queue 隊列名稱
21. * @param message 消息內容
22. * @param times 延遲時間 單位毫秒
23. */
24. public void send(String queueName,String message,long times);
25. }

[java] view plain copy

1. package com.ks.server.service.impl.queue;
3. import org.springframework.amqp.AmqpException;
4. import org.springframework.amqp.core.Message;
5. import org.springframework.amqp.core.MessagePostProcessor;
6. import org.springframework.amqp.rabbit.core.RabbitTemplate;
7. import org.springframework.beans.factory.annotation.Autowired;
8. import org.springframework.stereotype.Service;
10. import com.base.common.codec.JSONUtils;
11. import com.ks.common.constant.MQConstant;
12. import com.ks.common.service.queue.IMessageQueueService;
13. import com.ks.modal.queue.DLXMessage;
15. /**
16. *
17. * @author victor
18. * @desc 消息隊列服務接口實現
19. */
20. @Service("messageQueueService")
21. public class MessageQueueServiceImpl implements IMessageQueueService{
23. @Autowired
24. private RabbitTemplate rabbitTemplate;
26. @Override
27. public void send(String queueName, String msg) {
28. rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,queueName, msg);
29. }
31. @Override
32. public void send(String queueName, String msg, long times) {
33. DLXMessage dlxMessage = new DLXMessage(queueName,msg,times);
34. MessagePostProcessor processor = new MessagePostProcessor(){
35. @Override
36. public Message postProcessMessage(Message message) throws AmqpException {
37. message.getMessageProperties().setExpiration(times + "");
38. return message;
39. }
40. };
41. dlxMessage.setExchange(MQConstant.DEFAULT_EXCHANGE);
42. rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME, JSONUtils.toJson(dlxMessage), processor);
43. }
46. }

JSONUtils 為一個JSON工具類

新增消息實體，用於包裝消息：

[java] view plain copy

1. package com.ks.modal.queue;
3. import java.io.Serializable;
5. /**
6. *
7. * @author victor
8. * @desc rabbit 死信消息載體
9. */
10. public class DLXMessage implements Serializable {
12. private static final long serialVersionUID = 9956432152000L;
14. public DLXMessage() {
15. super();
16. }
18. public DLXMessage(String queueName, String content, long times) {
19. super();
20. this.queueName = queueName;
21. this.content = content;
22. this.times = times;
23. }
25. public DLXMessage(String exchange, String queueName, String content, long times) {
26. super();
27. this.exchange = exchange;
28. this.queueName = queueName;
29. this.content = content;
30. this.times = times;
31. }
34. private String exchange;
36. private String queueName;
38. private String content;
40. private long times;
42. //省略getter setter
44. }

路由轉發隊列消費者實現，負責接收超時消息，進行轉發：

[java] view plain copy

1. package com.ks.ons.processor.system;
3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;
4. import org.springframework.amqp.rabbit.annotation.RabbitListener;
5. import org.springframework.beans.factory.annotation.Autowired;
6. import org.springframework.stereotype.Component;
8. import com.base.common.codec.JSONUtils;
9. import com.ks.common.constant.MQConstant;
10. import com.ks.common.service.queue.IMessageQueueService;
11. import com.ks.modal.queue.DLXMessage;
13. /**
14. *
15. * @author victor
16. * @desc 死信接收處理消費者
17. */
18. @Component
19. @RabbitListener(queues = MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME)
20. public class TradeProcessor {
22. @Autowired
23. private IMessageQueueService messageQueueService;
25. @RabbitHandler
26. public void process(String content) {
27. DLXMessage message = JSONUtils.toBean(content, DLXMessage.class);
28. messageQueueService.send(message.getQueueName(), message.getContent());
29. }
30. }

啟動項目之後，管理界面如下:

測試代碼片段：

[java] view plain copy

1. messageQueueService.send(MQConstant.HELLO_QUEUE_NAME,"測試延遲發送消息",60000);

spring boot Rabbitmq集成,延時消息隊列實現