简单记录Spring项目集成RabbitMQ的过程,重点记录生产者项目自动创建队列的操作,因该问题给项目带来很多麻烦。
本文内容分别在Spring(V5.2.6)和Spring Boot(V2.5.14)两个项目中经过了验证,下述示例代码来自于SpringBoot项目,迁移到Spring项目中需稍微调整。
一、Spring Boot集成RabbitMQ
1. 在Maven中加入依赖
Spring项目和SpringBoot项目的依赖有区别,按需引入:
<!-- Spring Boot --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><!-- Spring --><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit</artifactId><version>2.2.9.RELEASE</version></dependency>
2. 在Spring配置文件增加配置项
spring:rabbitmq:# 基础项host: 192.168.1.123
port:5672username: admin
password: admin
# virtualhost需要提前在MQ的Web管理界面里手动创建,或者配置默认host"/"virtual-host: /test
# 生产者#确认消息已发送到交换机(Exchange)publisher-confirm-type: correlated
#确认消息已发送到队列(Queue)publisher-returns:true# 消费者listener:type: simple
simple:default-requeue-rejected:falseacknowledge-mode: auto #确认模式prefetch:1#限制每次发送一条数据max-concurrency:1#启动消费者最大数量concurrency:1#同一个队列启动几个消费者retry:enabled:true#是否支持重试
说明:这里仅按照现有项目的配置列出,在实际的项目中,还是需要根据自身实际情况做出调整。
3. 编写生产者代码
3.1 创建RabbitMQ配置类:
@ConfigurationpublicclassRabbitMQConfig{//queuepublicstaticfinalStringWORK_QUEUE="test.queue";//exchangepublicstaticfinalStringWORK_DIRECTEXCHANGE="test.directExchange";//routingpublicstaticfinalStringWORK_DIRECTROUTING="test.directRouting";@BeanpublicRabbitTemplaterabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate =newRabbitTemplate(connectionFactory);return rabbitTemplate;}// Queue@BeanpublicQueuedirectQueue(){Map<String,Object> argsMap =newHashMap<String,Object>();
argsMap.put("x-max-priority",5);Queue queue =newQueue(WORK_QUEUE,true,false,false, argsMap);return queue;}//Direct交换机@BeanDirectExchangedirectExchange(){returnnewDirectExchange(WORK_DIRECTEXCHANGE,true,false);}//绑定@BeanBindingbindingDirect(){returnBindingBuilder.bind(directQueue()).to(directExchange()).with(WORK_DIRECTROUTING);}}
3.2 编写发送消息工具类:
@ComponentpublicclassRabbitMQUtils{privatestaticRabbitTemplate rabbitTemplate;@AutowiredpublicRabbitMQUtils(RabbitTemplate rabbitTemplate){this.rabbitTemplate = rabbitTemplate;}publicstaticvoidsendMsg(String msg)throwsAmqpException{try{
rabbitTemplate.convertAndSend(RabbitMQConfig.WORK_DIRECTEXCHANGE,RabbitMQConfig.WORK_DIRECTROUTING, msg);}catch(AmqpException e){thrownewAmqpException("RabbitMQ发送消息异常", e);}}}
3.3 编写单元测试,测试发送消息结果。
4. 编写消费者代码
4.1 编写监听器:
@ComponentpublicclassRabbitMQListener{@RabbitListener(queues ={Constants.WORK_QUEUE})//监听队列publicvoidlistener(String msg,Message message){System.out.println(msg);System.out.println(message.getBody());}}
二、存在的问题
1. 问题描述
实际项目中,消息的生产者和消费者不在同一项目中,如果先启动消费者会因为没有队列而启动失败。
2. 尝试解决
2.1 方式一:最容易想到的是,在MQ的Web管理界面中手动创建队列:
- 该方式在实际操作是个不容易的事情,因为还要创建Channel和Exchange,何况发布的人不一定是开发的人,沟通繁琐,极易出错,好像程序还是半成品似的;
2.2 方式二:启动消费者项目时,监听器发现不存在队列自动创建:
- “通过@RabbitListener”的参数,确实可以实现;
- 但是这种方式在我的项目中出现了新问题,消费者项目启动后创建了队列,但是生产者发送消息出错,貌似没有了权限?
- 当时因为项目工期,并未深究,具体错误也没记录下来;应该是自己的代码有问题;
- 因此,不能确认该方式的可行性,或者具体实现方式。
2.3 方式三:先启动生产者项目:
- 因RabbitMQ懒加载模式,所以单纯启动项目是不会创建队列的;
- 因此,最开始的想法是,启动项目后,先发送一条测试消息去创建队列,项目确实用该方式使用了一段时间;
- 近期在升级项目时,发现个现象,没有队列时候的第一条消息确实可以创建队列,但是MQ里没有消息,是空队列;
- 种种问题促使自己重新寻找解决方式,在生产者项目启动后可以自动创建队列,因此有了本篇文章;
- 很惭愧,其实答案一直在,但是自己对MQ的认知一直停留在简单使用里,希望有机会能深入的学习一下吧。
三、生产者项目创建队列
1. 在RabbitMQ配置类中加入RabbitAdmin
@BeanpublicRabbitAdminrabbitAdmin(ConnectionFactory connectionFactory){RabbitAdmin rabbitAdmin =newRabbitAdmin(connectionFactory);
rabbitAdmin.setAutoStartup(true);return rabbitAdmin;}
2. 通过RabbitAdmin声明队列,完成队列的创建
@BeanpublicQueuedirectQueue(){Map<String,Object> argsMap =newHashMap<String,Object>();
argsMap.put("x-max-priority",5);Queue queue =newQueue(WORK_QUEUE,true,false,false, argsMap);// 声明队列
rabbitAdmin.declareQueue(queue);return queue;}
3. 修改后的完整配置类
@ConfigurationpublicclassRabbitMQConfig{//queuepublicstaticfinalStringWORK_QUEUE="test.queue";//exchangepublicstaticfinalStringWORK_DIRECTEXCHANGE="test.directExchange";//routingpublicstaticfinalStringWORK_DIRECTROUTING="test.directRouting";@AutowiredprivateRabbitAdmin rabbitAdmin;@BeanpublicRabbitTemplaterabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate =newRabbitTemplate(connectionFactory);return rabbitTemplate;}@BeanpublicRabbitAdminrabbitAdmin(ConnectionFactory connectionFactory){RabbitAdmin rabbitAdmin =newRabbitAdmin(connectionFactory);
rabbitAdmin.setAutoStartup(true);return rabbitAdmin;}@BeanpublicQueuedirectQueue(){Map<String,Object> argsMap =newHashMap<String,Object>();
argsMap.put("x-max-priority",5);Queue queue =newQueue(WORK_QUEUE,true,false,false, argsMap);
rabbitAdmin.declareQueue(queue);return queue;}//Direct交换机@BeanDirectExchangedirectExchange(){returnnewDirectExchange(WORK_DIRECTEXCHANGE,true,false);}//绑定@BeanBindingbindingDirect(){returnBindingBuilder.bind(directQueue()).to(directExchange()).with(WORK_DIRECTROUTING);}}
4. 启动项目,查看队列创建情况
- 经过多次测试,在仅手动创建virtual-host的前提下,启动项目,队列可以自动创建,且发送/接收消息都正常完成。
版权归原作者 清风xhp 所有, 如有侵权,请联系我们删除。