Rabbit MQ 消息队列
@[toc]
简介
优点
erlang 开发, 并发能力强.
社区活跃, 使用的人多, 稳定性较强.
延时低
缺点
erlang 语言开发的, 国内精通的不多, 日后定制开发困难.
Rabbit MQ 工作模式
1,"Hello World!" 模式
简单模式是 Rabbit MQ 最简单入门的案例, 其中一个生产者, 一个消费者. 无需声明交换机(其实是有个默认的交换机), 声明完一个队列以后, 生产者往这里发送, 消费者则监听该队列, 对消息做出响应.
应用场景: 如两个用户之间的聊天.
2,Work queues 模式
工作队列模式在工人之间分配任务(竞争的消费者模式), 就是有多个消费者的简单模式, 多个消费者一起监听该队列, Rabbit MQ 会确保队列中的某一条消息只会被某一个消费者消费.
应用场景: 如多个用户抢某个订单(高并发下争抢同一个资源的记得考虑加锁)
3,Publish/Subscribe 模式
发布 / 订阅模式一次向许多消费者发送消息, 还是一个生产者多个消费者, 但是该模式需要显示声明交换机了, 将创建的多个队列与该交换机进行绑定, 生产者将消息发给交换机, 由交换机发给已经绑定好的队列, 对应的某队列的消费者消费该队列中的信息.
应用场景: 群发某个邮件, 推送某条广告
4,Routing 模式
路由模式是有选择地接收消息, 还是显示声明交换机, 将队列与交换机绑定, 由生产者将消息发给交换机, 由交换机发给对应的队列, 由对应的消费者消费信息. 但是队列绑定交换机时需要指定一个 routingkey , 这样的话, 生产者再向生产者发消息的时候, 会判断具体的 routingkey 值, 并将符合对应 routingkey 值的消息发送给对应的队列.
应用场景: 封装若干类错误类型通知
5,Topics 模式
主题模式根据模式 (主题) 接收消息, 同路由模式一样, 只不过不再是根据具体的 routingkey 值做判断, 是根据 routingkey 值模糊匹配的. 其中, 星号代表多个单词 , 井号代表一个单词.
更加详细的资料可访问官网 : RabbitMQ 官网 https://www.rabbitmq.com/getstarted.html
交换机模式
fanout 模式: 不处理路由键, 即没有 routingkey 值, 很像子网广播, 每台子网内的主机都获得了一份复制的消息.
direct 模式: 处理路由键, 需要路由键匹配才能转发, 即完全匹配上 routingkey 值.
topic 模式: 处理路由键, 需要路由键模糊匹配才能转发, 即模糊匹配上 routingkey 值.
引入 RabbitMQ 队列
RabbitMQ 依赖与 erlang 语言, 点击 官网下载链接 http://erlang.org/download/ 下载 erlang.
点击 官网下载链接 https://www.rabbitmq.com/download.html 下载 RabbitMQ.
安装两者的时候, 一路默认设置即可.
启动 RabbitMQ 服务, cmd 模式下输入, 进入默认安装路径
C:\Program Files (x86)\RabbitMQ Server\rabbitmq_server-3.6.10\sbin
输入命令
rabbitmq-plugins enable rabbitmq_management
服务启动成功, 如图所示
点击 http://localhost:15672/ , 如下图所示
pom.xml 中新增
- <dependency>
- <groupId>
- org.springframework.boot
- </groupId>
- <artifactId>
- spring-boot-starter-amqp
- </artifactId>
- </dependency>
application.properties 中加入
- # RabbitMQ
- spring.rabbitmq.host=127.0.0.1
- spring.rabbitmq.port=5672
- spring.rabbitmq.username=guest
- spring.rabbitmq.password=guest
- # 并发消费者的初始化值
- spring.rabbitmq.listener.concurrency=10
- # 并发消费者的最大值
- spring.rabbitmq.listener.max-concurrency=20
- # 每个消费者每次监听时可拉取处理的消息数量
- spring.rabbitmq.listener.prefetch=5
代码实战
简单模式和工作队列模式没有指定交换机, rabbitMQ 的精妙之处不能体现出来, 而路由模式和主体模式都有 routingkey 值, 故挑其一, 本文以发布 / 订阅模式和路由模式为例, 其余的都可类比. 不全写, 是因为这是快速上手使用系列, 最短的时间精力明白来龙去脉, 才是该系列的宗旨, 人类畏惧晦涩冗长的, 喜欢简短直接的.
新增代码后的目录如下所示
RabbitConfig.java
- package com.example.config;
- import org.springframework.amqp.core.*;
- import org.springframework.context.annotation.Bean;
- import org.springframework.context.annotation.Configuration;
- @Configuration
- public class RabbitConfig {
- //============== 发布 / 订阅模式 ==============
- // 队列 A
- @Bean()
- public Queue queueA() {
- return new Queue("fanoutQueueA");
- }
- // 队列 B
- @Bean()
- public Queue queueB() {
- return new Queue("fanoutQueueB");
- }
- // 发布订阅模式下的 fanout 交换机
- @Bean
- public FanoutExchange fanoutExchange() {
- return new FanoutExchange("fanoutExchange");
- }
- // 把 队列 A 和 fanout 交换机 绑定在一起
- @Bean
- public Binding bindingWithQueueA() {
- return BindingBuilder.bind(queueA()).to(fanoutExchange());
- }
- // 把 队列 B 和 fanout 交换机 绑定在一起
- @Bean
- public Binding bindingWithQueueB() {
- return BindingBuilder.bind(queueB()).to(fanoutExchange());
- }
- //============== 路由模式 ==============
- // 队列 C
- @Bean()
- public Queue queueC() {
- return new Queue("directQueueC");
- }
- // 队列 D
- @Bean()
- public Queue queueD() {
- return new Queue("directQueueD");
- }
- // 发布订阅模式下的 direct 交换机
- @Bean
- public DirectExchange directExchange() {
- return new DirectExchange("directExchange");
- }
- // 把 队列 C 和 direct 交换机 绑定在一起
- @Bean
- public Binding bindingWithQueueC() {
- return BindingBuilder.bind(queueC()).to(directExchange()).with("directRoutingkey");
- }
- // 把 队列 B 和 direct 交换机 绑定在一起
- @Bean
- public Binding bindingWithQueueD() {
- return BindingBuilder.bind(queueD()).to(directExchange()).with("directRoutingkey");
- }
- }
RabbitController.java
- package com.example.controller;
- import com.example.service.IRabbitProducerService;
- import org.springframework.beans.factory.annotation.Autowired;
- import org.springframework.stereotype.Controller;
- import org.springframework.web.bind.annotation.RequestMapping;
- @Controller
- @RequestMapping("rabbit")
- public class RabbitController {
- @Autowired
- private IRabbitProducerService rabbitProducerService;
- @RequestMapping("/fanout")
- public void fanout() {
- for (int i = 0; i < 5; i++) {
- this.rabbitProducerService.producerFanout("发布订阅模式下的第" + i + "条信息");
- }
- }
- @RequestMapping("/direct")
- public void direct() {
- for (int i = 0; i < 5; i++) {
- this.rabbitProducerService.producerDirect("路由模式下的第" + i + "条信息");
- }
- }
- }
IRabbitProducerService.java
- package com.example.service;
- public interface IRabbitProducerService {
- void producerFanout(String message);
- void producerDirect(String message);
- }
RabbitProducerServiceIml.java
- package com.example.service;
- import org.springframework.amqp.rabbit.core.RabbitTemplate;
- import org.springframework.beans.factory.annotation.Autowired;
- import org.springframework.stereotype.Service;
- @Service("rabbitProducerService")
- public class RabbitProducerServiceIml implements IRabbitProducerService {
- @Autowired
- RabbitTemplate rabbitTemplate;
- public void producerFanout(String message) {
- rabbitTemplate.convertAndSend("fanoutExchange", null, message);
- }
- public void producerDirect(String message) {
- rabbitTemplate.convertAndSend("directExchange", "directRoutingkey", message);
- }
- }
IRabbitConsumerService.java
- package com.example.service;
- public interface IRabbitConsumerService {
- void consumerFanoutA(String message);
- void consumerFanoutB(String message);
- void consumerDirectC(String message);
- void consumerDirectD(String message);
- }
RabbitConsumerServiceIml.java
- package com.example.service;
- import org.springframework.amqp.rabbit.annotation.RabbitListener;
- import org.springframework.stereotype.Component;
- @Component
- public class RabbitConsumerServiceIml implements IRabbitConsumerService {
- @RabbitListener(queues = "fanoutQueueA")
- public void consumerFanoutA(String message) {
- System.out.println("消费者收到队列 A 中的消息:" + message);
- }
- @RabbitListener(queues = "fanoutQueueB")
- public void consumerFanoutB(String message) {
- System.out.println("消费者收到队列 B 中的消息:" + message);
- }
- @RabbitListener(queues = "routingQueueC")
- public void consumerDirectC(String message) {
- System.out.println("消费者收到队列 C 中的消息:" + message);
- }
- @RabbitListener(queues = "routingQueueD")
- public void consumerDirectD(String message) {
- System.out.println("消费者收到队列 D 中的消息:" + message);
- }
- }
启动项目后, 浏览器输入 http://localhost:8080/rabbit/fanout ,(虽然前台会报错, 因为这个请求只是触发消息的生产, 并没有对应的界面展示. 生产出来的消息经过交换机传给队列, 消费者则监听队列, 做出对应的响应.)发布 / 订阅模式下使用 fanout 交换机生产消费消息的效果图如下所示;
浏览器输入 http://localhost:8080/rabbit/direct , 发路由模式下使用 direct 交换机生产消费消息的效果图如下所示;
注: 如果是根据本文系列文章来的, 因为一开始就配置好了 spring security, 所以记得将该地址配给所登录的用户. 或者开一个超级管理员账号, 可以访问项目的任意目录, 使用该管理员账号访问这些地址.
到这里, 篇幅已经很长了, 本来想准备写完死信队列, 和消息确认机制. 想想这是快速整合上手系列, 就把深入的内容放到踩坑深入系列, 尽情期待后续开辟新的系列文章吧.
来源: http://blog.51cto.com/14089205/2454825