详解spring boot集成RabbitMQ
RabbitMQ作为AMQP的代表性产品，在项目中大量使用。结合现在主流的springboot，极大简化了开发过程中所涉及到的消息通信问题。
首先正确的安装RabbitMQ及运行正常。
RabbitMQ需啊erlang环境，所以首先安装对应版本的erlang，可在RabbitMQ官网下载
#rpm-ivherlang-19.0.4-1.el7.centos.x86_64.rpm
使用yum安装RabbitMQ,避免缺少依赖包引起的安装失败
#yuminstallrabbitmq-server-3.6.6-1.el7.noarch.rpm
启动RabbitMQ
#/sbin/servicerabbitmq-serverstart
由于RabbitMQ默认提供的guest用户只能本地访问，所以额外创建用户用于测试
#/sbin/rabbitmqctladd_usertesttest123 用户名：test，密码：test123
开启web管理插件
#rabbitmq-pluginsenablerabbitmq_management
并使用之前创建的用户登录,并设置该用户为administrator,虚拟主机地址为/
springboot引入相关依赖
org.springframework.boot spring-boot-starter-amqp
消息生产者
application.properties添加一下配置
spring.rabbitmq.host=192.168.1.107 spring.rabbitmq.port=5672 spring.rabbitmq.username=test spring.rabbitmq.password=test123 spring.rabbitmq.publisher-confirms=true spring.rabbitmq.publisher-returns=true spring.rabbitmq.template.mandatory=true
springboot配置类，作用为指定队列，交换器类型及绑定操作
importorg.springframework.amqp.core.Binding;
importorg.springframework.amqp.core.BindingBuilder;
importorg.springframework.amqp.core.Queue;
importorg.springframework.amqp.core.TopicExchange;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
@Configuration
publicclassRabbitConfig{
//声明队列
@Bean
publicQueuequeue1(){
returnnewQueue("hello.queue1",true);//true表示持久化该队列
}
@Bean
publicQueuequeue2(){
returnnewQueue("hello.queue2",true);
}
//声明交互器
@Bean
TopicExchangetopicExchange(){
returnnewTopicExchange("topicExchange");
}
//绑定
@Bean
publicBindingbinding1(){
returnBindingBuilder.bind(queue1()).to(topicExchange()).with("key.1");
}
@Bean
publicBindingbinding2(){
returnBindingBuilder.bind(queue2()).to(topicExchange()).with("key.#");
}
}
共声明了2个队列，分别是hello.queue1，hello.queue2，交换器类型为TopicExchange,并与hello.queue1，hello.queue2队列分别绑定。
生产者类
importjava.util.UUID;
importjavax.annotation.PostConstruct;
importorg.springframework.amqp.core.Message;
importorg.springframework.amqp.rabbit.core.RabbitTemplate;
importorg.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
importorg.springframework.amqp.rabbit.support.CorrelationData;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.stereotype.Component;
@Component
publicclassSenderimplementsRabbitTemplate.ConfirmCallback,ReturnCallback{
@Autowired
privateRabbitTemplaterabbitTemplate;
@PostConstruct
publicvoidinit(){
rabbitTemplate.setConfirmCallback(this);
rabbitTemplate.setReturnCallback(this);
}
@Override
publicvoidconfirm(CorrelationDatacorrelationData,booleanack,Stringcause){
if(ack){
System.out.println("消息发送成功:"+correlationData);
}else{
System.out.println("消息发送失败:"+cause);
}
}
@Override
publicvoidreturnedMessage(Messagemessage,intreplyCode,StringreplyText,Stringexchange,StringroutingKey){
System.out.println(message.getMessageProperties().getCorrelationIdString()+"发送失败");
}
//发送消息，不需要实现任何接口，供外部调用。
publicvoidsend(Stringmsg){
CorrelationDatacorrelationId=newCorrelationData(UUID.randomUUID().toString());
System.out.println("开始发送消息:"+msg.toLowerCase());
Stringresponse=rabbitTemplate.convertSendAndReceive("topicExchange","key.1",msg,correlationId).toString();
System.out.println("结束发送消息:"+msg.toLowerCase());
System.out.println("消费者响应:"+response+"消息处理完成");
}
}
要点：
1.注入RabbitTemplate
2.实现RabbitTemplate.ConfirmCallback,RabbitTemplate.ReturnCallback接口（非必须）。
ConfirmCallback接口用于实现消息发送到RabbitMQ交换器后接收ack回调。ReturnCallback接口用于实现消息发送到RabbitMQ交换器，但无相应队列与交换器绑定时的回调。
3.实现消息发送方法。调用rabbitTemplate相应的方法即可。rabbitTemplate常用发送方法有
rabbitTemplate.send(message);//发消息，参数类型为org.springframework.amqp.core.Message rabbitTemplate.convertAndSend(object);//转换并发送消息。将参数对象转换为org.springframework.amqp.core.Message后发送 rabbitTemplate.convertSendAndReceive(message)//转换并发送消息,且等待消息者返回响应消息。
针对业务场景选择合适的消息发送方式即可。
消息消费者
application.properties添加一下配置
spring.rabbitmq.host=192.168.1.107 spring.rabbitmq.port=5672 spring.rabbitmq.username=test spring.rabbitmq.password=test123 spring.rabbitmq.listener.concurrency=2//最小消息监听线程数 spring.rabbitmq.listener.max-concurrency=2//最大消息监听线程数
消费者类
importorg.springframework.amqp.rabbit.annotation.RabbitListener;
importorg.springframework.stereotype.Component;
@Component
publicclassReceiver{
@RabbitListener(queues="hello.queue1")
publicStringprocessMessage1(Stringmsg){
System.out.println(Thread.currentThread().getName()+"接收到来自hello.queue1队列的消息："+msg);
returnmsg.toUpperCase();
}
@RabbitListener(queues="hello.queue2")
publicvoidprocessMessage2(Stringmsg){
System.out.println(Thread.currentThread().getName()+"接收到来自hello.queue2队列的消息："+msg);
}
}
由于定义了2个队列，所以分别定义不同的监听器监听不同的队列。由于最小消息监听线程数和最大消息监听线程数都是2，所以每个监听器各有2个线程实现监听功能。
要点:
1.监听器参数类型与消息实际类型匹配。在生产者中发送的消息实际类型是String，所以这里监听器参数类型也是String。
2.如果监听器需要有响应返回给生产者，直接在监听方法中return即可。
运行测试
importjava.util.Date;
importorg.junit.Test;
importorg.junit.runner.RunWith;
importorg.springframework.beans.factory.annotation.Autowired;
importorg.springframework.boot.test.context.SpringBootTest;
importorg.springframework.test.context.junit4.SpringJUnit4ClassRunner;
importcom.sam.demo.rabbitmq.Application;
importcom.sam.demo.rabbitmq.sender.Sender;
@RunWith(value=SpringJUnit4ClassRunner.class)
@SpringBootTest(classes=Application.class)
publicclassRabbitTests{
@Autowired
privateSendersender;
@Test
publicvoidsendTest()throwsException{
while(true){
Stringmsg=newDate().toString();
sender.send(msg);
Thread.sleep(1000);
}
}
}
输出:
开始发送消息:wedmar2923:20:52cst2017 SimpleAsyncTaskExecutor-1接收到来自hello.queue2队列的消息：WedMar2923:20:52CST2017 SimpleAsyncTaskExecutor-2接收到来自hello.queue1队列的消息：WedMar2923:20:52CST2017 结束发送消息:wedmar2923:20:52cst2017 消费者响应:WEDMAR2923:20:52CST2017消息处理完成 ------------------------------------------------ 消息发送成功:CorrelationData[id=340d14e6-cfcc-4653-9f95-29b37d50f886] 开始发送消息:wedmar2923:20:53cst2017 SimpleAsyncTaskExecutor-1接收到来自hello.queue1队列的消息：WedMar2923:20:53CST2017 SimpleAsyncTaskExecutor-2接收到来自hello.queue2队列的消息：WedMar2923:20:53CST2017 结束发送消息:wedmar2923:20:53cst2017 消费者响应:WEDMAR2923:20:53CST2017消息处理完成 ------------------------------------------------ 消息发送成功:CorrelationData[id=e4e01f89-d0d4-405e-80f0-85bb20238f34] 开始发送消息:wedmar2923:20:54cst2017 SimpleAsyncTaskExecutor-2接收到来自hello.queue1队列的消息：WedMar2923:20:54CST2017 SimpleAsyncTaskExecutor-1接收到来自hello.queue2队列的消息：WedMar2923:20:54CST2017 结束发送消息:wedmar2923:20:54cst2017 消费者响应:WEDMAR2923:20:54CST2017消息处理完成 ------------------------------------------------
如果需要使用的其他的交换器类型，spring中都已提供实现，所有的交换器均实现org.springframework.amqp.core.AbstractExchange接口。
常用交换器类型如下：
Direct(DirectExchange)：direct类型的行为是"先匹配,再投送".即在绑定时设定一个routing_key,消息的routing_key完全匹配时,才会被交换器投送到绑定的队列中去。
Topic(TopicExchange)：按规则转发消息（最灵活）。
Headers(HeadersExchange)：设置headerattribute参数类型的交换机。
Fanout(FanoutExchange)：转发消息到所有绑定队列。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。