初窥RabbitMQ消息中间及SpringBoot整合

2次阅读

共计 8825 个字符，预计需要花费 23 分钟才能阅读完成。

【本文版权归微信公众号 ” 代码艺术 ”（ID：onblog）所有，若是转载请务必保留本段原创声明，违者必究。若是文章有不足之处，欢迎关注微信公众号私信与我进行交流！】

RabbitMQ 即一个消息队列，主要是用来实现应用程序的异步和解耦，同时也能起到消息缓冲，消息分发的作用。

消息中间件最主要的作用是解耦，中间件最标准的用法是生产者生产消息传送到队列，消费者从队列中拿取消息并处理，生产者不用关心是谁来消费，消费者不用关心谁在生产消息，从而达到解耦的目的。在分布式的系统中，消息队列也会被用在很多其它的方面，比如：分布式事务的支持，RPC 的调用等等。

RabbitMQ 是实现 AMQP（高级消息队列协议）的消息中间件的一种，最初起源于金融系统，用于在分布式系统中存储转发消息，在易用性、扩展性、高可用性等方面表现不俗。RabbitMQ 主要是为了实现系统之间的双向解耦而实现的。当生产者大量产生数据时，消费者无法快速消费，那么需要一个中间层。保存这个数据。

AMQP，即 Advanced Message Queuing Protocol，高级消息队列协议，是应用层协议的一个开放标准，为面向消息的中间件设计。消息中间件主要用于组件之间的解耦，消息的发送者无需知道消息使用者的存在，反之亦然。AMQP 的主要特征是面向消息、队列、路由（包括点对点和发布 / 订阅）、可靠性、安全。

RabbitMQ 是一个开源的 AMQP 实现，服务器端用 Erlang 语言编写，支持多种客户端，如：Python、Ruby、.NET、Java、JMS、C、PHP、ActionScript、XMPP、STOMP 等，支持 AJAX。

通常我们谈到队列服务, 会有三个概念：发消息者、队列、收消息者，RabbitMQ 在这个基本概念之上, 多做了一层抽象, 在发消息者和队列之间, 加入了交换器 (Exchange). 这样发消息者和队列就没有直接联系, 转而变成发消息者把消息给交换器, 交换器根据调度策略再把消息再给队列。

左侧 P 代表生产者，也就是往 RabbitMQ 发消息的程序。
中间即是 RabbitMQ，其中包括了交换机和队列。
右侧 C 代表消费者，也就是往 RabbitMQ 拿消息的程序。

那么，其中比较重要的概念有 4 个，分别为：虚拟主机，交换机，队列，和绑定。

虚拟主机：一个虚拟主机持有一组交换机、队列和绑定。为什么需要多个虚拟主机呢？很简单，RabbitMQ 当中，用户只能在虚拟主机的粒度进行权限控制。 因此，如果需要禁止 A 组访问 B 组的交换机 / 队列 / 绑定，必须为 A 和 B 分别创建一个虚拟主机。每一个 RabbitMQ 服务器都有一个默认的虚拟主机“/”。
交换机：Exchange 用于转发消息，但是它不会做存储 ，如果没有 Queue bind 到 Exchange 的话，它会直接丢弃掉 Producer 发送过来的消息。这里有一个比较重要的概念： 路由键（routing_key）。消息到交换机的时候，交互机会转发到对应的队列中，那么究竟转发到哪个队列，就要根据该路由键。
绑定：也就是交换机需要和队列相绑定，这其中如上图所示，是多对多的关系。

交换机的功能主要是接收消息并且转发到绑定的队列，交换机不存储消息，在启用 ack 模式后，交换机找不到队列会返回错误。交换机有四种类型：Direct, topic, Headers and Fanout

Direct：direct 类型的行为是”先匹配, 再投送”. 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.
Topic：按规则转发消息（最灵活）
Headers：设置 header attribute 参数类型的交换机
Fanout：转发消息到所有绑定队列（广播：忽略 routing_key）

RabbitMQ 简单来说，就是生产者发送消息到虚拟主机，虚拟主机把消息交给指定的交换机，交换机按照规则扔给消息队列进行存储，消息队列等待消费者来消费。

由此我想到了商品买卖：厂家生产商品卖给批发部，批发部交给指定的超市出售，超市按照售价摆放在门店，并等待顾客上门购买。

完美~~

因为 RabbitMQ 是由 erlang 语言写的，就像 Java 程序需要 jdk 环境一样，运行 RabbitMQ 也需要 erlang 环境。

环境：Centos7.4

下载地址：http://erlang.org/download/

在 Linux 终端运行命令行

下载：

wget http://erlang.org/download/otp_src_18.3.tar.gz

下载一定要认准 otp_src_ 字样。

解压：

tar -zxvf otp_src_18.3.tar.gz

进入解压后的根目录：

./configure --prefix=/usr/local/erlang --enable-hipe --enable-threads --enable-smp-support --enable-kernel-poll --without-javac

make && make install

上面有点慢。

把 erlang 加入环境变量：

vi /etc/profile

export ERLANG=/usr/local/erlang/erlang
export PATH=$ERLANG/bin:$PATH

使环境变量生效

source /etc/profile

然后，我们测试下是否安装成功：

[root@ystcode ~]# erl
Erlang/OTP 18 [erts-7.3]  [64-bit] [async-threads:10] [hipe] [kernel-poll:false]

Eshell V7.3  (abort with ^G)
1>

下载地址：http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/

下载：

wget http://www.rabbitmq.com/releases/rabbitmq-server/v3.6.1/rabbitmq-server-generic-unix-3.6.1.tar.xz

对于下载 xz 包进行解压，首先先下载 xz 压缩工具：

yum install xz

对 rabbitmq 包进行解压：

xz -d xz -d rabbitmq-server-generic-unix-3.6.1.tar.xz

tar -xvf rabbitmq-server-generic-unix-3.6.1.tar

随后在 sbin 目录启用 MQ 管理方式：

./rabbitmq-plugins enable rabbitmq_management   #启动后台管理

./rabbitmq-server -detached #后台运行 rabbitmq

添加用户和权限默认网页 guest 用户是不允许访问的，需要增加一个用户修改一下权限，代码如下：

添加用户:

./rabbitmqctl add_user admin admin

添加权限:

./rabbitmqctl set_permissions -p "/" admin ".*" ".*" ".*"

修改用户角色:

./rabbitmqctl set_user_tags admin administrator

然后就可以远程访问了，然后可直接配置用户权限等信息。

验证

访问 http://localhost:15672/，输入 admin 用户密码，登录成功！

新建一个 1.5 版本的 Spring boot 项目，选择 rabbitmq+web 模块。

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

virtual-host: / 默认就是斜杠，具体查看 rabbitmq 后台 admin 栏。如果默认这行不用写。

spring:
  rabbitmq:
    host: 127.0.0.1
    username: admin
    password: admin
    virtual-host: / #可以不用写

direct 类型的行为是”先匹配, 再投送”. 即在绑定时设定一个 routing_key, 消息的routing_key 匹配时, 才会被交换器投送到绑定的队列中去.

1. 我们在 RabbitMQ 后台新建一个交换机，demo-direct交换机名，direct交换机类型，Durable持久化：

2. 再新建一个消息队列，取名为demo：

3. 点击 demo-direct 交换机进入绑定消息队列demo：

如果没有指定 routingkey，消息队列的名称就是 routingkey

4. 绑定成功后查看：

你可以直接在交换机页面下方的 Publish message 发送消息，在消息队列的 Get message 查看消息，不过我们实际生产比较多使用编程：

在 Spring Boot 创建单元测试

demo-direct交换机名

demo是该交换机绑定的消息队列名

发送消息

@Autowired
RabbitTemplate rabbitTemplate;

@Test
public void contextLoads() {
   //message 需要自己构造一个，定义消息体内容和消息体
   //rabbitTemplate.send(exchange,routingkey,message);
   Map map = new HashMap();
   map.put("key","值");
   map.put("msg",true);
   // 对象被默认序列化后发送
   rabbitTemplate.convertAndSend("demo-direct","demo",map);
}

此时发送消息我们在 rabbitmq 网页发现消息是经序列化后的，我们如果想改变序列化机制为 JSON，也很简单，只需要注入一个人 Bean：

@Configuration
public class MyAMQPConfig {
    @Bean
    public MessageConverter messageConverter(){return new Jackson2JsonMessageConverter();
    }
}

然后我们再启动测试，发送。

接收消息

demo是消息队列名，也就是消费者只需要得到消息队列的名字就可以接收队列中的消息。

@Test
public void receive(){Object o = rabbitTemplate.receiveAndConvert("demo");
    System.out.println(o.getClass());
    System.out.println(o);
}

查看打印

class java.util.HashMap
{msg=true, key= 值}

【本文版权归微信公众号 ” 代码艺术 ”（ID：onblog）所有，若是转载请务必保留本段原创声明，违者必究。若是文章有不足之处，欢迎关注微信公众号私信与我进行交流！】

转发消息到所有绑定队列（广播：忽略 routing_key）

1. 首先我们需要在 RabbitMQ 后台创建一个广播交换机：

2. 再创建一些（demo1，demo2）消息队列，以一个作为演示：

3. 进入创建的交换机页面：

4. 交换机与消息队列（demo1,demo2）进行绑定：

如果没有指定 routingkey，消息队列的名称就是 routingkey

在 Spring Boot 创建单元测试

注意在广播模式下会忽略忽略 routing_key

发送消息

@Autowired
RabbitTemplate rabbitTemplate;

@Test
public void send() {Book book = new Book();
    book.setName("< 西游记 >");
    book.setAnthony("吴承恩");
    // 对象被默认序列化后发送
    rabbitTemplate.convertAndSend("demo-fanout","",book);
}

接收消息

@Test
public void receive(){Object o = rabbitTemplate.receiveAndConvert("demo");
    System.out.println(o.getClass());
    System.out.println(o);
}

打印输出：

class cn.zyzpp.rabbitmq.entity.Book
Book{name='< 西游记 >', anthony='吴承恩'}

按规则转发消息（使用通配符）

1. 在 rabbitmq 后台新建一个 topic 交换机

2. 新建一个消息队列demo

3. 交换机绑定消息队列，注意此处的 路由键 Routing key使用了通配符

*表示一个词.
#表示零个或多个词.

那我们如何区分几个字母为一个单词呢？

答案是通过”点分”的 routing_key 形式，比如两个单词是 *.demo hello.demo，

如果路由键为 demo.#, 那可以匹配demo. 开头的所有路由键。

4. 查看此时的交换机

在 Spring Boot 创建单元测试

发送消息

@Test
public void contextLoads() {
    //message 需要自己构造一个，定义消息体内容和消息体
    //rabbitTemplate.send(exchange,routingkey,message);
    Map map = new HashMap();
    map.put("key","topic 交换机");
    map.put("msg",true);
    // 对象被默认序列化后发送
    rabbitTemplate.convertAndSend("demo-topic","demo.hello",map);
}

接收消息

@Test
public void receive(){Object o = rabbitTemplate.receiveAndConvert("demo");
    System.out.println(o.getClass());
    System.out.println(o);
}

打印输出

class java.util.HashMap
{msg=true, key=topic 交换机}

1. 上面演示的是通过 RabbitMQ 网页后台创建，通过编程的方式也非常简单：

@Autowired
AmqpAdmin amqpAdmin;

/**
 * 代码创建交换机与消息队列并绑定
 */
@Test
public void createExChange(){new TopicExchange("topic.exChange");
      new FanoutExchange("fanout.exChange");
    amqpAdmin.declareExchange(new DirectExchange("amqp.exChange"));// 创建交换机（remove 为删除交换机）System.out.println("单播交换机创建完成");
    amqpAdmin.declareQueue(new Queue("amqp.queue",true)); // 创建消息队列
    amqpAdmin.declareBinding(new Binding("amqp.queue",
            Binding.DestinationType.QUEUE,"amqp.exChange","amqp.exChange",null));// 绑定
}

2. 登录后台查看，创建成功！

监听消息队列，当有消息发到消息队列，立马获取并操作。方法也很简单。

1. 开启基于注解的 Rabbit 模式

@EnableRabbit   // 开启基于注解的 Rabbit 模式
@SpringBootApplication
public class RabbitmqApplication {public static void main(String[] args) {SpringApplication.run(RabbitmqApplication.class, args);
    }
}

2.@RabbitListener 注解实现监听

注意此时你的消息队列已经有了 demo 和demo.news

@Service
public class BookService {@RabbitListener(queues = "demo")
    public void receive(Book book){System.out.println("收到消息："+ book);
    }

    @RabbitListener(queues = "demo.news")
    public void receiveMess(Message message){System.out.println("收到消息："+ message);
        System.out.println(message.getBody()); //getBody 返回的 byte[]字节数组
        System.out.println(message.getMessageProperties());
    }
}

然后我们在开启主程序时再运行测试用例：

测试用例

@Test
public void send() {Book book = new Book();
    book.setName("< 西游记 >");
    book.setAnthony("吴承恩");
    // 对象被默认序列化后发送
    rabbitTemplate.convertAndSend("demo-fanout","",book);
}

查看主控制台打印

收到消息：Book{name='< 西游记 >', anthony='吴承恩'}

测试用例

@Test
public void contextLoads() {
    //message 需要自己构造一个，定义消息体内容和消息体
    //rabbitTemplate.send(exchange,routingkey,message);
    Map map = new HashMap();
    map.put("key","topic 交换机");
    map.put("msg",true);
    // 对象被默认序列化后发送
    rabbitTemplate.convertAndSend("demo-topic","demo.hello",map);
}

查看主控制台打印

收到消息：(Body:'{"msg":true,"key":"topic 交换机 "}' MessageProperties [headers={__ContentTypeId__=java.lang.Object, __KeyTypeId__=java.lang.Object, __TypeId__=java.util.HashMap}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/json, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=demo-topic, receivedRoutingKey=demo.hello, receivedDelay=null, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-KRZh2DyNETjEaGSH0JZ2dA, consumerQueue=demo.news])
[B@5332f99e
MessageProperties [headers={__ContentTypeId__=java.lang.Object, __KeyTypeId__=java.lang.Object, __TypeId__=java.util.HashMap}, timestamp=null, messageId=null, userId=null, receivedUserId=null, appId=null, clusterId=null, type=null, correlationId=null, correlationIdString=null, replyTo=null, contentType=application/json, contentEncoding=UTF-8, contentLength=0, deliveryMode=null, receivedDeliveryMode=PERSISTENT, expiration=null, priority=0, redelivered=false, receivedExchange=demo-topic, receivedRoutingKey=demo.hello, receivedDelay=null, deliveryTag=1, messageCount=0, consumerTag=amq.ctag-KRZh2DyNETjEaGSH0JZ2dA, consumerQueue=demo.news]

【本文版权归微信公众号 ” 代码艺术 ”（ID：onblog）所有，若是转载请务必保留本段原创声明，违者必究。若是文章有不足之处，欢迎关注微信公众号私信与我进行交流！】

正文完

程序员

发表至：程序员

2020-06-02

0

关于程序员:3本书阅读笔记人月神话Go语言实战研发能力持续成长路线01

关于程序员:Mummer-用法简析

关于程序员:直播视频源码开发技术之平台测试之软件测试侧重点介绍

关于程序员:大咖说实在智能｜RPA上云加速电商数字员工时代的到来

HyperfCrontab-组件源码解析

初窥RabbitMQ消息中间及SpringBoot整合

一：RabbitMQ 简介

RabbitMQ 介绍

运行机制

交换机

一句话总结

二：RabbitMQ 安装

1. 安装 erlang

2. 安装 RabbitMQ

三：整合 Spring Boot

1. 导入依赖

2.yml 配置 rabbitmq

3. 发送 direct 消息

4. 广播 fanout 消息

5. 发送 topic 消息

6. 代码创建交换机与消息队列

7. 监听消息队列

版权声明

站内搜索