Spring boot集成RabbitMQ的示例代码

2023-09-12 15:41:04 255

RabbitMQ简介

RabbitMQ是一个在AMQP基础上完整的，可复用的企业消息系统

MQ全称为MessageQueue,消息队列（MQ）是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息（针对应用程序的数据）来通信，而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信，而不是通过直接调用彼此来通信，直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

AMQP就是一个协议，是一个高级抽象层消息通信协议。

虽然在同步消息通讯的世界里有很多公开标准（如COBAR的IIOP，或者是SOAP等），但是在异步消息处理中却不是这样，只有大企业有一些商业实现（如微软的MSMQ，IBM的WebsphereMQ等），因此，在2006年的6月，Cisco、Redhat、iMatix等联合制定了AMQP的公开标准。也就是说AMQP是异步通讯的一个协议。

RabbitMQ使用场景

在项目中，将一些无需即时返回且耗时的操作提取出来，进行了异步处理，而这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量。不过大多数不仅仅是无需即时返回，甚至是执行是否成功都无所谓。如果需要即时返回则可以使用Dubbo,Springboot与Dubbo集成可以去看Springboot集成Dubbox

RabbitMQ依赖

RabbitMQ并不是直接一个简单的jar包（Jar包只是提供一个基本的与RabbitMQ本身通讯的一些功能），和Dubbo相同，RabbitMQ也需要其他软件来运行，以下是RabbitMQ运行所需要的软件

1、Erlang

由于RabbitMQ软件本身是基于Erlang开发的，所以想要运行RabbitMQ必须要先按照Erlang

Erlang官网

Erlang下载地址

RabbitMQ

RabbitMQ才是实现消息队列的核心

RabbitMQ官网

RabbitMQ下载

配置RabbitMQ

安装完成后，需要完成一些配置才能使用RabbitMQ，可以直接用cmd到RabbitMQ的安装目录下的sbin目录通过命令配置，也可以直接在开始菜单中直接找到RabbitMQCommandPrompt(sbindir)运行直接到达RabbitMQ的安装目录的sbin，为了方便，我们先启用管理插件,执行命令

rabbitmq-plugins.batenablerabbitmq_management

即可，注意，这是在Windows下面，如果是Linux则没有bat后缀然后我们添加一个用户，因为在外网环境没有用户的情况下是不能连接成功的，执行添加用户命令

rabbitmqctl.batadd_userspringbootpassword

springboot是用户名，password是密码

然后为了方便演示，我们给springboot赋予管理员权限，方便登录管理页面

rabbitmqctl.batset_user_tagsspringbootadministrator

给账号赋予虚拟主机权限

rabbitmqctl.batset_permissions-p/springboot.*.*.*

然后启动RabbitMQ服务访问RabbitMQ管理页面http://localhost:15672即可看见登录页面，如果没有创建用户则可以用guest，guest登录，如果有创建用户则用创建的用户登录

创建Springboot项目

因为创建springboot项目在前面的文章已经说过很多次了，所以这里就不多说了

添加RabbitMQ相关依赖



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

没错，就是点配置，不过这样可能有点不理解，我还是把全部配置贴出来吧


4.0.0

wang.raye.rabbitmq
demo1
0.0.1-SNAPSHOT
jar

demo1
http://maven.apache.org


UTF-8


org.springframework.boot
spring-boot-starter-parent
1.4.0.RELEASE




junit
junit
3.8.1
test



org.springframework.boot
spring-boot-starter-web




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

因为没有做其他操作，所以目前项目主要是依赖2个模块，一个Sprigboot，一个RabbitMQ

添加配置类

packagewang.raye.rabbitmq.demo1;
importorg.springframework.amqp.core.AcknowledgeMode;
importorg.springframework.amqp.core.Binding;
importorg.springframework.amqp.core.BindingBuilder;
importorg.springframework.amqp.core.DirectExchange;
importorg.springframework.amqp.core.Message;
importorg.springframework.amqp.core.Queue;
importorg.springframework.amqp.rabbit.connection.CachingConnectionFactory;
importorg.springframework.amqp.rabbit.connection.ConnectionFactory;
importorg.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
importorg.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
importorg.springframework.context.annotation.Bean;
importorg.springframework.context.annotation.Configuration;
/**
*rabbitmq的配置类
*
*@authorRaye
*@since2016年10月12日10:57:44
*/
@Configuration
publicclassRabbitMQConfig{
/**消息交换机的名字*/
publicstaticfinalStringEXCHANGE="my-mq-exchange";
/**队列key1*/
publicstaticfinalStringROUTINGKEY1="queue_one_key1";
/**队列key2*/
publicstaticfinalStringROUTINGKEY2="queue_one_key2";

/**
*配置链接信息
*@return
*/
@Bean
publicConnectionFactoryconnectionFactory(){
CachingConnectionFactoryconnectionFactory=newCachingConnectionFactory("127.0.0.1",5672);

connectionFactory.setUsername("springboot");
connectionFactory.setPassword("password");
connectionFactory.setVirtualHost("/");
connectionFactory.setPublisherConfirms(true);//必须要设置
returnconnectionFactory;
}

/**
*配置消息交换机
*针对消费者配置
FanoutExchange:将消息分发到所有的绑定队列，无routingkey的概念
HeadersExchange：通过添加属性key-value匹配
DirectExchange:按照routingkey分发到指定队列
TopicExchange:多关键字匹配
*/
@Bean
publicDirectExchangedefaultExchange(){
returnnewDirectExchange(EXCHANGE,true,false);
}
/**
*配置消息队列1
*针对消费者配置
*@return
*/
@Bean
publicQueuequeue(){
returnnewQueue("queue_one",true);//队列持久

}
/**
*将消息队列1与交换机绑定
*针对消费者配置
*@return
*/
@Bean
publicBindingbinding(){
returnBindingBuilder.bind(queue()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY1);
}

/**
*配置消息队列2
*针对消费者配置
*@return
*/
@Bean
publicQueuequeue1(){
returnnewQueue("queue_one1",true);//队列持久

}
/**
*将消息队列2与交换机绑定
*针对消费者配置
*@return
*/
@Bean
publicBindingbinding1(){
returnBindingBuilder.bind(queue1()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY2);
}
/**
*接受消息的监听，这个监听会接受消息队列1的消息
*针对消费者配置
*@return
*/
@Bean
publicSimpleMessageListenerContainermessageContainer(){
SimpleMessageListenerContainercontainer=newSimpleMessageListenerContainer(connectionFactory());
container.setQueues(queue());
container.setExposeListenerChannel(true);
container.setMaxConcurrentConsumers(1);
container.setConcurrentConsumers(1);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL);//设置确认模式手工确认
container.setMessageListener(newChannelAwareMessageListener(){
publicvoidonMessage(Messagemessage,com.rabbitmq.client.Channelchannel)throwsException{
byte[]body=message.getBody();
System.out.println("收到消息:"+newString(body));
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);//确认消息成功消费

}

});
returncontainer;
}
/**
*接受消息的监听，这个监听会接受消息队列1的消息
*针对消费者配置
*@return
*/
@Bean
publicSimpleMessageListenerContainermessageContainer2(){
SimpleMessageListenerContainercontainer=newSimpleMessageListenerContainer(connectionFactory());
container.setQueues(queue1());
container.setExposeListenerChannel(true);
container.setMaxConcurrentConsumers(1);
container.setConcurrentConsumers(1);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL);//设置确认模式手工确认
container.setMessageListener(newChannelAwareMessageListener(){

publicvoidonMessage(Messagemessage,com.rabbitmq.client.Channelchannel)throwsException{
byte[]body=message.getBody();
System.out.println("queue1收到消息:"+newString(body));
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);//确认消息成功消费
}
});
returncontainer;
}
}

注意，为了更好的展示如何配置，我配置了2个消息队列，而本类除了链接配置哪里，其他都是针对消息消费者的，当然不管消息消费者和消息生产者都需要配置链接信息，而为了方便，所以本项目的消息消费者和生产者都在本项目，一般实际项目中不会在同一项目，由于注释很详细，我就不多说了

发送消息

为了方便发送消息，所以我直接写了一个Controller，通过访问接口的形式来调用发送消息的方法，话不多说，上代码

packagewang.raye.rabbitmq.demo1;
importjava.util.UUID;
importorg.springframework.amqp.rabbit.core.RabbitTemplate;
importorg.springframework.amqp.rabbit.support.CorrelationData;
importorg.springframework.web.bind.annotation.RequestMapping;
importorg.springframework.web.bind.annotation.RestController;

/**
*测试RabbitMQ发送消息的Controller
*@authorRaye
*
*/
@RestController
publicclassSendControllerimplementsRabbitTemplate.ConfirmCallback{
privateRabbitTemplaterabbitTemplate;
/**
*配置发送消息的rabbitTemplate，因为是构造方法，所以不用注解Spring也会自动注入（应该是新版本的特性）
*@paramrabbitTemplate
*/
publicSendController(RabbitTemplaterabbitTemplate){
this.rabbitTemplate=rabbitTemplate;
//设置消费回调
this.rabbitTemplate.setConfirmCallback(this);
}
/**
*向消息队列1中发送消息
*@parammsg
*@return
*/
@RequestMapping("send1")
publicStringsend1(Stringmsg){
Stringuuid=UUID.randomUUID().toString();
CorrelationDatacorrelationId=newCorrelationData(uuid);
rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE,RabbitMQConfig.ROUTINGKEY1,msg,
correlationId);
returnnull;
}
/**
*向消息队列2中发送消息
*@parammsg
*@return
*/
@RequestMapping("send2")
publicStringsend2(Stringmsg){
Stringuuid=UUID.randomUUID().toString();
CorrelationDatacorrelationId=newCorrelationData(uuid);
rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE,RabbitMQConfig.ROUTINGKEY2,msg,
correlationId);
returnnull;
}
/**
*消息的回调，主要是实现RabbitTemplate.ConfirmCallback接口
*注意，消息回调只能代表成功消息发送到RabbitMQ服务器，不能代表消息被成功处理和接受
*/
publicvoidconfirm(CorrelationDatacorrelationData,booleanack,Stringcause){
System.out.println("回调id:"+correlationData);
if(ack){
System.out.println("消息成功消费");
}else{
System.out.println("消息消费失败:"+cause+"\n重新发送");
}
}
}

需要注意的是消息回调只能代表消息成功发送到RabbitMQ服务器

然后我们启动项目，访问http://localhost:8082/send1?msg=aaaa会发现控制台输出了

收到消息:aaaa
回调id:CorrelationData[id=37e6e913-835a-4eca-98d1-807325c5900f]
消息成功消费

当然回调id可能不同，如果我们访问http://localhost:8082/send2?msg=bbbb则输出

queue1收到消息:bbbb
回调id:CorrelationData[id=0cec7500-3117-4aa2-9ea5-4790879812d4]
消息成功消费

最后说两句

因为本文主要是说明如何从零到springboot集成RabbitMQ，所以对于RabbitMQ的很多信息和用法没有说明，如果对RabbitMQ本身不太熟悉的可以去看看其他关于RabbitMQ的文章,附上本文demo

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。

Spring boot集成RabbitMQ的示例代码

热门推荐

随机推荐