700字范文,内容丰富有趣,生活中的好帮手!
700字范文 > Spring boot集成RabbitMQ

Spring boot集成RabbitMQ

时间:2019-09-05 23:43:33

相关推荐

Spring boot集成RabbitMQ

####RabbitMQ简介 RabbitMQ是一个在AMQP基础上完整的,可复用的企业消息系统 MQ全称为Message Queue, 消息队列(MQ)是一种应用程序对应用程序的通信方法。应用程序通过读写出入队列的消息(针对应用程序的数据)来通信,而无需专用连接来链接它们。消息传递指的是程序之间通过在消息中发送数据进行通信,而不是通过直接调用彼此来通信,直接调用通常是用于诸如远程过程调用的技术。排队指的是应用程序通过 队列来通信。队列的使用除去了接收和发送应用程序同时执行的要求。

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

虽然在同步消息通讯的世界里有很多公开标准(如 COBAR的 IIOP ,或者是 SOAP 等),但是在异步消息处理中却不是这样,只有大企业有一些商业实现(如微软的 MSMQ ,IBM 的 Websphere MQ 等),因此,在 年的 6 月,Cisco 、Redhat、iMatix 等联合制定了 AMQP 的公开标准。也就是说AMQP是异步通讯的一个协议。 ####RabbitMQ使用场景 在项目中,将一些无需即时返回且耗时的操作提取出来,进行了异步处理,而这种异步处理的方式大大的节省了服务器的请求响应时间,从而提高了系统的吞吐量。不过大多数不仅仅是无需即时返回,甚至是执行是否成功都无所谓。如果需要即时返回则可以使用Dubbo,Spring boot与Dubbo集成可以去看Spring boot 集成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目录通过命令配置,也可以直接在开始菜单中直接找到RabbitMQ Command Prompt (sbin dir)运行直接到达RabbitMQ的安装目录的sbin,为了方便,我们先启用管理插件,执行命令

rabbitmq-plugins.bat enable rabbitmq_management

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

rabbitmqctl.bat add_user springboot password

springboot是用户名,password是密码 然后为了方便演示,我们给springboot赋予管理员权限,方便登录管理页面

rabbitmqctl.bat set_user_tags springboot administrator

给账号赋予虚拟主机权限

rabbitmqctl.bat set_permissions -p / springboot .* .* .*

然后启动RabbitMQ服务 访问RabbitMQ管理页面http://localhost:15672即可看见登录页面,如果没有创建用户则可以用guest,guest登录,如果有创建用户则用创建的用户登录 ####创建Springboot项目 因为创建spring boot项目在前面的文章已经说过很多次了,所以这里就不多说了,如果不会可以去看其他关于spring boot的博客 ####添加RabbitMQ相关依赖

<!-- rabbitmq --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency>

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

<project xmlns="/POM/4.0.0" xmlns:xsi="/2001/XMLSchema-instance"xsi:schemaLocation="/POM/4.0.0 /xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>wang.raye.rabbitmq</groupId><artifactId>demo1</artifactId><version>0.0.1-SNAPSHOT</version><packaging>jar</packaging><name>demo1</name><url></url><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>1.4.0.RELEASE</version></parent><dependencies><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>3.8.1</version><scope>test</scope></dependency><!-- Springboot --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- rabbitmq --><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency></dependencies></project>

因为没有做其他操作,所以目前项目主要是依赖2个模块,一个Sprig boot,一个RabbitMQ ####添加配置类

package wang.raye.rabbitmq.demo1;import org.springframework.amqp.core.AcknowledgeMode;import org.springframework.amqp.core.Binding;import org.springframework.amqp.core.BindingBuilder;import org.springframework.amqp.core.DirectExchange;import org.springframework.amqp.core.Message;import org.springframework.amqp.core.Queue;import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;import org.springframework.amqp.rabbit.connection.ConnectionFactory;import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;/*** rabbitmq 的配置类* * @author Raye* @since 10月12日10:57:44*/@Configuration public class RabbitMQConfig {/** 消息交换机的名字*/public static final String EXCHANGE = "my-mq-exchange";/** 队列key1*/public static final String ROUTINGKEY1 = "queue_one_key1";/** 队列key2*/public static final String ROUTINGKEY2 = "queue_one_key2";/*** 配置链接信息* @return*/@Beanpublic ConnectionFactory connectionFactory() {CachingConnectionFactory connectionFactory = new CachingConnectionFactory("127.0.0.1",5672);connectionFactory.setUsername("springboot");connectionFactory.setPassword("password");connectionFactory.setVirtualHost("/");connectionFactory.setPublisherConfirms(true); // 必须要设置return connectionFactory;}/** * 配置消息交换机* 针对消费者配置 FanoutExchange: 将消息分发到所有的绑定队列,无routingkey的概念 HeadersExchange :通过添加属性key-value匹配 DirectExchange:按照routingkey分发到指定队列 TopicExchange:多关键字匹配 */ @Bean public DirectExchange defaultExchange() { return new DirectExchange(EXCHANGE, true, false);} /*** 配置消息队列1* 针对消费者配置 * @return*/@Bean public Queue queue() { return new Queue("queue_one", true); //队列持久 }/*** 将消息队列1与交换机绑定* 针对消费者配置 * @return*/@Bean public Binding binding() { return BindingBuilder.bind(queue()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY1); } /*** 配置消息队列2* 针对消费者配置 * @return*/@Bean public Queue queue1() { return new Queue("queue_one1", true); //队列持久 }/*** 将消息队列2与交换机绑定* 针对消费者配置 * @return*/@Bean public Binding binding1() { return BindingBuilder.bind(queue1()).to(defaultExchange()).with(RabbitMQConfig.ROUTINGKEY2); } /*** 接受消息的监听,这个监听会接受消息队列1的消息* 针对消费者配置 * @return*/@Bean public SimpleMessageListenerContainer messageContainer() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory()); container.setQueues(queue()); container.setExposeListenerChannel(true); container.setMaxConcurrentConsumers(1); container.setConcurrentConsumers(1); container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认 container.setMessageListener(new ChannelAwareMessageListener() {public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {byte[] body = message.getBody(); System.out.println("收到消息 : " + new String(body)); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费 } }); return container; } /*** 接受消息的监听,这个监听会接受消息队列1的消息* 针对消费者配置 * @return*/@Bean public SimpleMessageListenerContainer messageContainer2() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory()); container.setQueues(queue1()); container.setExposeListenerChannel(true); container.setMaxConcurrentConsumers(1); container.setConcurrentConsumers(1); container.setAcknowledgeMode(AcknowledgeMode.MANUAL); //设置确认模式手工确认 container.setMessageListener(new ChannelAwareMessageListener() {public void onMessage(Message message, com.rabbitmq.client.Channel channel) throws Exception {byte[] body = message.getBody(); System.out.println("queue1 收到消息 : " + new String(body)); channel.basicAck(message.getMessageProperties().getDeliveryTag(), false); //确认消息成功消费 } }); return container; } }

注意,为了更好的展示如何配置,我配置了2个消息队列,而本类除了链接配置哪里,其他都是针对消息消费者的,当然不管消息消费者和消息生产者都需要配置链接信息,而为了方便,所以本项目的消息消费者和生产者都在本项目,一般实际项目中不会在同一项目,由于注释很详细,我就不多说了 ####发送消息 为了方便发送消息,所以我直接写了一个Controller,通过访问接口的形式来调用发送消息的方法,话不多说,上代码

package wang.raye.rabbitmq.demo1;import java.util.UUID;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.amqp.rabbit.support.CorrelationData;import org.springframework.web.bind.annotation.RequestMapping;import org.springframework.web.bind.annotation.RestController;/*** 测试RabbitMQ发送消息的Controller* @author Raye**/@RestControllerpublic class SendController implements RabbitTemplate.ConfirmCallback{private RabbitTemplate rabbitTemplate;/*** 配置发送消息的rabbitTemplate,因为是构造方法,所以不用注解Spring也会自动注入(应该是新版本的特性)* @param rabbitTemplate*/public SendController(RabbitTemplate rabbitTemplate){this.rabbitTemplate = rabbitTemplate;//设置消费回调this.rabbitTemplate.setConfirmCallback(this);}/*** 向消息队列1中发送消息* @param msg* @return*/@RequestMapping("send1")public String send1(String msg){String uuid = UUID.randomUUID().toString();CorrelationData correlationId = new CorrelationData(uuid);rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTINGKEY1, msg,correlationId);return null;}/*** 向消息队列2中发送消息* @param msg* @return*/@RequestMapping("send2")public String send2(String msg){String uuid = UUID.randomUUID().toString();CorrelationData correlationId = new CorrelationData(uuid);rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTINGKEY2, msg,correlationId);return null;}/*** 消息的回调,主要是实现RabbitTemplate.ConfirmCallback接口* 注意,消息回调只能代表成功消息发送到RabbitMQ服务器,不能代表消息被成功处理和接受*/public void confirm(CorrelationData correlationData, boolean ack, String cause) {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

本内容不代表本网观点和政治立场,如有侵犯你的权益请联系我们处理。
网友评论
网友评论仅供其表达个人看法,并不表明网站立场。