首页 > 编程语言 > 详细

spring boot Rabbitmq集成,延时消息队列实现

时间：2017-09-21 11:01:02      阅读：498      评论：0      收藏：0      [点我收藏+]
本篇主要记录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 {  
  4.   
  5.     private static Logger logger = Logger.getLogger(RabbitMQConfiguration.class);  
  6.   
  7.     private String host;  
  8.   
  9.     private int port;  
  10.   
  11.     private String username;  
  12.   
  13.     private String password;  
  14.   
  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.     }  
  26.   
  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 {  
  7.       
  8.     /** 
  9.      * 发送消息到队列 
  10.      * @param queue 队列名称 
  11.      * @param message 消息内容 
  12.      */  
  13.     public void send(String queueName,String message);  
  14.       
  15.       
  16. }  

 

 

Service实现

 

[java] view plain copy
 
  1. package com.ks.server.service.impl.queue;  
  2.   
  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;  
  9.   
  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;  
  14.   
  15. /** 
  16.  *  
  17.  * @author victor 
  18.  * @desc 消息队列服务接口实现 
  19.  */  
  20. @Service("messageQueueService")  
  21. public class MessageQueueServiceImpl implements IMessageQueueService{  
  22.       
  23.     @Autowired  
  24.     private RabbitTemplate rabbitTemplate;  
  25.   
  26.     @Override  
  27.     public void send(String queueName, String msg) {  
  28.         rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,queueName, msg);  
  29.     }  
  30.   
  31.   
  32.   
  33. }  


相关常量类:

 

 

[java] view plain copy
 
  1. package com.ks.common.constant;  
  2.   
  3. /** 
  4.  *  
  5.  * @author victor 
  6.  * @desc Rabbit消息队列相关常量 
  7.  */  
  8. public final class MQConstant {  
  9.       
  10.     private MQConstant(){  
  11.     }  
  12.       
  13.     //exchange name  
  14.     public static final String DEFAULT_EXCHANGE = "KSHOP";  
  15.       
  16.     //DLX QUEUE  
  17.     public static final String DEFAULT_DEAD_LETTER_QUEUE_NAME = "kshop.dead.letter.queue";  
  18.       
  19.     //DLX repeat QUEUE 死信转发队列  
  20.     public static final String DEFAULT_REPEAT_TRADE_QUEUE_NAME = "kshop.repeat.trade.queue";  
  21.       
  22.       
  23.     //Hello 测试消息队列名称  
  24.     public static final String HELLO_QUEUE_NAME = "HELLO";  
  25.       
  26.       
  27. }  

 

 

到现在为止,队列相关配置，以及使用以及封装完成，接下来是创建队列，

这里我是单独创建一个配置类，用于队列配置， 创建Hello队列示例如下:

 

[java] view plain copy
 
  1. package com.ks.ons.config;  
  2.   
  3. import java.util.HashMap;  
  4. import java.util.Map;  
  5.   
  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;  
  12.   
  13. import com.ks.common.constant.MQConstant;  
  14.   
  15. /** 
  16.  *  
  17.  * @author victor 
  18.  * @desc 队列配置 
  19.  */  
  20. @Configuration  
  21. public class QueueConfiguration {  
  22.       
  23.     //信道配置  
  24.     @Bean  
  25.     public DirectExchange defaultExchange() {  
  26.         return new DirectExchange(MQConstant.DEFAULT_EXCHANGE, true, false);  
  27.     }  
  28.   
  29.       
  30.     /*********************    hello 队列  测试    *****************/  
  31.     @Bean  
  32.     public Queue queue() {  
  33.         Queue queue = new Queue(MQConstant.HELLO_QUEUE_NAME,true);  
  34.         return queue;   
  35.     }  
  36.   
  37.     @Bean  
  38.     public Binding binding() {  
  39.         return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.HELLO_QUEUE_NAME);  
  40.     }  
  41.   
  42. }  


通过配置队列bean，在程序启动时会在rabbitmq中创建相关队列，启动程序，可以在rabbtmq管理界面看到信道和队列信息:

 

技术分享

技术分享

 

 

众所周知，既然有了队列，用来处理业务的最终还是需要消费者，消费者创建示例如下:

 

[java] view plain copy
 
  1. package com.ks.ons.processor.hello;  
  2.   
  3. import org.springframework.amqp.rabbit.annotation.RabbitHandler;  
  4. import org.springframework.amqp.rabbit.annotation.RabbitListener;  
  5. import org.springframework.stereotype.Component;  
  6.   
  7. import com.ks.common.constant.MQConstant;  
  8.   
  9. /** 
  10.  *  
  11.  * @author victor 
  12.  * @desc hello 消息队列消费者 
  13.  */  
  14. @Component  
  15. @RabbitListener(queues = MQConstant.HELLO_QUEUE_NAME)  
  16. public class HelloProcessor {  
  17.       
  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 {  
  8.       
  9.     //信道配置  
  10.     @Bean  
  11.     public DirectExchange defaultExchange() {  
  12.         return new DirectExchange(MQConstant.DEFAULT_EXCHANGE, true, false);  
  13.     }  
  14.   
  15.       
  16.     @Bean  
  17.     public Queue repeatTradeQueue() {  
  18.         Queue queue = new Queue(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME,true,false,false);  
  19.         return queue;   
  20.     }  
  21.       
  22.     @Bean  
  23.     public Binding  drepeatTradeBinding() {  
  24.         return BindingBuilder.bind(repeatTradeQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME);  
  25.     }  
  26.   
  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.     }  
  36.   
  37.     @Bean  
  38.     public Binding  deadLetterBinding() {  
  39.         return BindingBuilder.bind(deadLetterQueue()).to(defaultExchange()).with(MQConstant.DEFAULT_DEAD_LETTER_QUEUE_NAME);  
  40.     }  
  41.       
  42.       
  43.       
  44.     /*********************    hello 队列  测试    *****************/  
  45.     @Bean  
  46.     public Queue queue() {  
  47.         Queue queue = new Queue(MQConstant.HELLO_QUEUE_NAME,true);  
  48.         return queue;   
  49.     }  
  50.   
  51.     @Bean  
  52.     public Binding binding() {  
  53.         return BindingBuilder.bind(queue()).to(defaultExchange()).with(MQConstant.HELLO_QUEUE_NAME);  
  54.     }  
  55.   
  56. }  


修改Service服务：

 

[java] view plain copy
 
  1. package com.ks.common.service.queue;  
  2.   
  3.   
  4. /** 
  5.  *  
  6.  * @author victor 
  7.  * @desc 消息队列服务接口 
  8.  */  
  9. public interface IMessageQueueService {  
  10.       
  11.     /** 
  12.      * 发送消息到队列 
  13.      * @param queue 队列名称 
  14.      * @param message 消息内容 
  15.      */  
  16.     public void send(String queueName,String message);  
  17.       
  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;  
  2.   
  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;  
  9.   
  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;  
  14.   
  15. /** 
  16.  *  
  17.  * @author victor 
  18.  * @desc 消息队列服务接口实现 
  19.  */  
  20. @Service("messageQueueService")  
  21. public class MessageQueueServiceImpl implements IMessageQueueService{  
  22.       
  23.     @Autowired  
  24.     private RabbitTemplate rabbitTemplate;  
  25.   
  26.     @Override  
  27.     public void send(String queueName, String msg) {  
  28.         rabbitTemplate.convertAndSend(MQConstant.DEFAULT_EXCHANGE,queueName, msg);  
  29.     }  
  30.   
  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.     }  
  44.   
  45.   
  46. }  

 

JSONUtils 为一个JSON工具类

 

 

新增消息实体，用于包装消息：

 

[java] view plain copy
 
  1. package com.ks.modal.queue;  
  2.   
  3. import java.io.Serializable;  
  4.   
  5. /** 
  6.  *  
  7.  * @author victor 
  8.  * @desc rabbit 死信消息载体 
  9.  */  
  10. public class DLXMessage implements Serializable {  
  11.   
  12.     private static final long serialVersionUID = 9956432152000L;  
  13.       
  14.     public DLXMessage() {  
  15.         super();  
  16.     }  
  17.   
  18.     public DLXMessage(String queueName, String content, long times) {  
  19.         super();  
  20.         this.queueName = queueName;  
  21.         this.content = content;  
  22.         this.times = times;  
  23.     }  
  24.   
  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.     }  
  32.   
  33.   
  34.     private String exchange;  
  35.       
  36.     private String queueName;  
  37.       
  38.     private String content;  
  39.       
  40.     private long times;  
  41.   
  42.     //省略getter setter  
  43.       
  44. }  

 

 

路由转发队列消费者实现，负责接收超时消息，进行转发：

 

[java] view plain copy
 
  1. package com.ks.ons.processor.system;  
  2.   
  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;  
  7.   
  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;  
  12.   
  13. /** 
  14.  *  
  15.  * @author victor 
  16.  * @desc 死信接收处理消费者 
  17.  */  
  18. @Component  
  19. @RabbitListener(queues = MQConstant.DEFAULT_REPEAT_TRADE_QUEUE_NAME)  
  20. public class TradeProcessor {  
  21.       
  22.     @Autowired  
  23.     private IMessageQueueService messageQueueService;  
  24.   
  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集成,延时消息队列实现

原文：http://www.cnblogs.com/vianzhang/p/7567328.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
分享档案
更多>
2021年09月23日 (328)
2021年09月24日 (313)
2021年09月17日 (191)
2021年09月15日 (369)
2021年09月16日 (411)
2021年09月13日 (439)
2021年09月11日 (398)
2021年09月12日 (393)
2021年09月10日 (160)
2021年09月08日 (222)
最新文章
更多>
教程昨日排行
更多>
友情链接
汇智网   PHP教程   插件网  
关于我们 - 联系我们 - 留言反馈 - 联系我们:wmxa8@hotmail.com
© 2014 bubuko.com 版权所有
打开技术之扣，分享程序人生！