RabbitMQ笔记_基础篇

  • MQ基本概念
    • 1. MQ概述
    • 2. MQ的优势和劣势
      • 2.1 优势☆
      • 2.2 劣势
      • 2.3 使用 MQ 需要满足什么条件呢?
    • 3. 常见的MQ产品
  • RabbitMQ基本介绍
    • 1. RabbitMQ 基础架构
    • 2. RabbitMQ 中的相关概念
    • 3. RabbitMQ的6 种工作模式☆
    • 4. AMQP 和 JMS
      • 4.1 AMQP
      • 4.2 JMS
      • 4.3 AMQP 与 JMS 区别☆
      • 4.4 再谈市场上常见的消息队列
  • AMQP
    • 1. 相关概念介绍
    • 2. RabbitMQ运转流程
    • 3. 生产者流转过程说明☆
    • 4. 消费者流转过程说明☆
  • RabbitMQ的安装和配置(了解)
    • 1. 安装依赖环境
    • 2. 安装Erlang
      • 2.1升级glibc(非必须)
    • 3. 安装RabbitMQ
    • 4. 开启管理界面及配置
    • 5. 启动
    • 6. 配置虚拟主机及用户
      • 6.1 用户角色
      • 6.2 Virtual Hosts配置
  • RabbitMQ入门
      • 1. 搭建示例工程
      • 2. 小结
  • RabbitMQ工作模式
    • 1. Helloworld简单模式
    • 2. Work queues工作队列模式
      • 2.1 模式说明
      • 2.2 代码
      • 2.3 测试
      • 2.4 小结
    • 3. Publish/Subscribe发布订阅模式☆
      • 3.1 模式说明
      • 3.2. 代码
      • 3.3 测试
      • 3.4 小结
    • 4. Routing路由模式
      • 4.1 模式说明
      • 4.2 代码
      • 4.3 测试
      • 4.4 小结
    • 5. Topics通配符模式
      • 5.1. 模式说明
      • 5.2. 代码
      • 5.3 测试
      • 5.4. 小结
    • 6. 工作模式总结
  • Spring 整合RabbitMQ
    • 1. 搭建工程
    • 2. 消息监听器
        • 队列监听器SpringQueueListener
        • 广播监听器1 FanoutListener1
        • 广播监听器2FanoutListener2
        • 星号通配符监听器TopicListenerStar
        • 井号通配符监听器TopicListenerWell
        • 井号通配符监听器2TopicListenerWell2
    • 3. 测试
  • Spring Boot整合RabbitMQ
    • 1. 简介
    • 2. 搭建生产者工程
      • 2.1 创建工程
      • 2.2 添加依赖
      • 2.3 启动类
      • 2.4 配置RabbitMQ
        • 配置文件
        • 绑定交换机和队列
    • 3. 搭建消费者工程
      • 3.1 创建工程
      • 3.2 添加依赖
      • 3.3 启动类
      • 3.4 配置RabbitMQ
      • 3.5 消息监听处理类
    • 4. 测试

MQ基本概念

1. MQ概述

MQ全称 Message Queue(消息队列),是在消息传输过程中保存消息的容器。用于分布式系统之间进行通信。

使用MQ中间件

小结:

  1. MQ—消息队列,存储消息的中间件。
  2. 分布式系统通信两种方式:直接远程调用借助第三方完成间接通信
  3. 发送方称为生产者,接收方称为消费者。

2. MQ的优势和劣势

2.1 优势☆

应用解耦 提升容错性和可维护性

系统的耦合性越高,容错性就越低,可维护性就越低。

使用 MQ 使得应用间解耦,提升容错性和可维护性。


异步提速 提升用户体验和系统吞吐量(单位时间内处理请求的数目)。

一个下单操作耗时:20 + 300 + 300 + 300 = 920ms,用户点击完下单按钮后,需要等待920ms才能得到下单响应,太慢!

用户点击完下单按钮后,只需等待20 + 5 = 25ms就能得到下单响应 。


削峰填谷 可以提高系统稳定性

使用了 MQ 之后,限制消费消息的速度为1000,这样一来,高峰期产生的数据势必会被积压在 MQ 中,高峰就被“削”掉了,但是因为消息积压,在高峰期过后的一段时间内,消费消息的速度还是会维持在1000,直到消费完积压的消息,这就叫做“填谷”

2.2 劣势

系统可用性降低

系统引入的外部依赖越多,系统稳定性越差。一旦 MQ 宕机,就会对业务造成影响。如何保证MQ的高可用?

系统复杂度提高

MQ 的加入大大增加了系统的复杂度,以前系统间是同步的远程调用,现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费?怎么处理消息丢失情况?那么保证消息传递的顺序性?

一致性问题

A 系统处理完业务,通过 MQ 给B、C、D三个系统发消息数据,如果 B 系统、C 系统处理成功,D 系统处理失败。如何保证消息数据处理的一致性?

2.3 使用 MQ 需要满足什么条件呢?

  1. 生产者不需要从消费者处获得反馈。

引入消息队列之前的直接调用,其接口的返回值应该为空,这才让明明下层的动作还没做,上层却当成动作做完了继续往后走,即所谓异步成为了可能。

  1. 容许短暂的不一致性。
  2. 解耦、提速、削峰这些方面的收益,超过加入MQ,管理MQ这些成本。

3. 常见的MQ产品

RabbitMQ基本介绍

2007年,Rabbit 技术公司基于 AMQP 标准开发的 RabbitMQ 1.0 发布。RabbitMQ 采用 Erlang 语言开发。Erlang 语言由 Ericson 设计,专门为开发高并发和分布式系统的一种语言,在电信领域使用广泛。

AMQP,即 Advanced Message Queuing Protocol(高级消息队列协议),是一个网络协议,是应用层协议的一个开放标准,为面向消息的中间件设计。基于此协议的客户端与消息中间件可传递消息,并不受客户端/中间件不同产品,不同的开发语言等条件的限制。2006年,AMQP 规范发布。类比HTTP。

1. RabbitMQ 基础架构

2. RabbitMQ 中的相关概念

Broker:接收和分发消息的应用,RabbitMQ Server就是 Message Broker。

Virtual host:出于多租户和安全因素设计的,把 AMQP 的基本组件划分到一个虚拟的分组中,类似于网络中的 namespace 概念。当多个不同的用户使用同一个 RabbitMQ server 提供的服务时,可以划分出多个vhost,每个用户在自己的 vhost 创建 exchange/queue 等。

Connection:publisher/consumer 和 broker 之间的 TCP 连接。

Channel:如果每一次访问 RabbitMQ 都建立一个 Connection,在消息量大的时候建立 TCP Connection的开销将是巨大的,效率也较低。Channel 是在 connection 内部建立的逻辑连接,如果应用程序支持多线程,通常每个thread创建单独的 channel 进行通讯,AMQP method 包含了channel id 帮助客户端和message broker 识别 channel,所以 channel 之间是完全隔离的。Channel 作为轻量级的 Connection 极大减少了操作系统建立 TCP connection 的开销。

Exchange:message 到达 broker 的第一站,根据分发规则,匹配查询表中的 routing key,分发消息到queue 中去。常用的类型有:

  • direct (point-to-point)
  • topic (publish-subscribe)
  • fanout (multicast)

Queue:消息最终被送到这里等待 consumer 取走

Binding:exchange 和 queue 之间的虚拟连接,binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中,用于 message 的分发依据

3. RabbitMQ的6 种工作模式☆

简单模式、work queues、Publish/Subscribe 发布与订阅模式、Routing 路由模式、Topics 主题模式、RPC 远程调用模式(远程调用,不太算 MQ;暂不作介绍)。

官网对应模式介绍:https://www.rabbitmq.com/getstarted.html

4. AMQP 和 JMS

MQ是消息通信的模型;实现MQ的大致有两种主流方式:AMQP、JMS。

4.1 AMQP

AMQP是一种协议,更准确的说是一种binary wire-level protocol(链接协议)。这是其和JMS的本质差别,AMQP不从API层进行限定,而是直接定义网络交换的数据格式

4.2 JMS

JMS即Java消息服务(JavaMessage Service)应用程序接口,是一个Java平台中关于面向消息中间件(MOM)的API,用于在两个应用程序之间,或分布式系统中发送消息,进行异步通信。

JMS 是 JavaEE 规范中的一种,类比JDBC。

4.3 AMQP 与 JMS 区别☆

本质:JMS是定义了统一的接口,对消息进行统一操作;AMQP只是协议,只规定了数据交互的格式

语言:JMS必须使用Java语言;AMQP不规定实现方式,是跨语言的。

消息模式:JMS规定了两种,而AMQP的多种消息模式

4.4 再谈市场上常见的消息队列

ActiveMQ:基于JMS

ZeroMQ:基于C语言开发

RabbitMQ:基于AMQP协议,erlang语言开发,稳定性好

RocketMQ:基于JMS,阿里巴巴产品

Kafka:类似MQ的产品;分布式消息系统,高吞吐量。


AMQP

1. 相关概念介绍

AMQP 一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。

AMQP是一个二进制协议,拥有一些现代化特点:多信道、协商式,异步,安全,扩平台,中立,高效。

RabbitMQ是AMQP协议的Erlang的实现。

概念说明
连接Connection一个网络连接,比如TCP/IP套接字连接。
会话Session端点之间的命名对话。在一个会话上下文中,保证“恰好传递一次”。
信道Channel多路复用连接中的一条独立的双向数据流通道。为会话提供物理传输介质。
客户端ClientAMQP连接或者会话的发起者。AMQP是非对称的,客户端生产和消费消息,服务器存储和路由这些消息。
服务节点Broker消息中间件的服务节点;一般情况下可以将一个RabbitMQ Broker看作一台RabbitMQ 服务器。
端点AMQP对话的任意一方。一个AMQP连接包括两个端点(一个是客户端,一个是服务器)。
消费者Consumer一个从消息队列里请求消息的客户端程序。
生产者Producer一个向交换机发布消息的客户端应用程序。

2. RabbitMQ运转流程

在入门案例中:

  • 生产者发送消息
    1. 生产者创建连接(Connection),开启一个信道(Channel),连接到RabbitMQ Broker;
    2. 声明队列并设置属性;是否排它,是否持久化,是否自动删除;
    3. 将路由键(空字符串)与队列绑定起来;
    4. 发送消息至RabbitMQ Broker;
    5. 关闭信道;
    6. 关闭连接;
  • 消费者接收消息
    1. 消费者创建连接(Connection),开启一个信道(Channel),连接到RabbitMQ Broker
    2. 向Broker 请求消费相应队列中的消息,设置相应的回调函数;
    3. 等待Broker回应闭关投递响应队列中的消息,消费者接收消息;
    4. 确认(ack,自动确认)接收到的消息;
    5. RabbitMQ从队列中删除相应已经被确认的消息;
    6. 关闭信道;
    7. 关闭连接;

3. 生产者流转过程说明☆

  1. 客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ,以此通知Broker 本次交互采用的是AMQPO-9-1 协议,紧接着Broker 返回Connection.Start 来建立连接,在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ,Connection.Open/ .Open-Ok 这6 个命令的交互。
  2. 客户端调用connection.createChannel方法。此方法开启信道,其包装的channel.open命令发送给Broker,等待channel.basicPublish方法,对应的AMQP命令为Basic.Publish,这个命令包含了content Header 和content Body()。content Header 包含了消息体的属性,例如:投递模式,优先级等,content Body 包含了消息体本身。
  3. 客户端发送完消息需要关闭资源时,涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。

4. 消费者流转过程说明☆

  1. 消费者客户端与代理服务器Broker建立连接。会调用newConnection() 方法,这个方法会进一步封装Protocol Header 0-9-1 的报文头发送给Broker ,以此通知Broker 本次交互采用的是AMQPO-9-1 协议,紧接着Broker 返回Connection.Start 来建立连接,在连接的过程中涉及Connection.Start/.Start-OK 、Connection.Tune/.Tune-Ok ,Connection.Open/ .Open-Ok 这6 个命令的交互。
  2. 消费者客户端调用connection.createChannel方法。和生产者客户端一样,协议涉及Channel . Open/Open-Ok命令。
  3. 在真正消费之前,消费者客户端需要向Broker 发送Basic.Consume 命令(即调用channel.basicConsume 方法〉将Channel 置为接收模式,之后Broker 回执Basic . Consume – Ok 以告诉消费者客户端准备好消费消息。
  4. Broker 向消费者客户端推送(Push) 消息,即Basic.Deliver 命令,这个命令和Basic.Publish 命令一样会携带Content Header 和Content Body。
  5. 消费者接收到消息并正确消费之后,向Broker 发送确认,即Basic.Ack 命令。
  6. 客户端发送完消息需要关闭资源时,涉及到Channel.Close和Channl.Close-Ok 与Connetion.Close和Connection.Close-Ok的命令交互。


RabbitMQ的安装和配置(了解)

https://blog.csdn.net/Kermit_father/article/details/105181740

1. 安装依赖环境

在线安装依赖环境:

yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c++ kernel-devel m4 ncurses-devel tk tc xz

2. 安装Erlang

下载下面三个安装包

erlang-18.3-1.el7.centos.x86_64.rpm

socat-1.7.3.2-5.el7.lux.x86_64.rpm

rabbitmq-server-3.6.5-1.noarch.rpm

#下载erlangwget www.rabbitmq.com/releases/erlang/erlang-18.3-1.el7.centos.x86_64.rpm#下载socatwget http://repo.iotti.biz/CentOS/7/x86_64/socat-1.7.3.2-5.el7.lux.x86_64.rpm#下载rabbitmq-serverwget www.rabbitmq.com/releases/rabbitmq-server/v3.6.5/rabbitmq-server-3.6.5-1.noarch.rpm# 安装rpm -ivh erlang-18.3-1.el7.centos.x86_64.rpm

如果出现如下错误

说明gblic 版本太低。我们可以查看当前机器的gblic 版本

strings /lib64/libc.so.6 | grep GLIBC

2.1升级glibc(非必须)

当前最高版本2.12,需要2.15.所以需要升级glibc

  • 使用yum更新安装依赖

    sudo yum install zlib-devel bzip2-devel openssl-devel ncurses-devel sqlite-devel readline-devel tk-devel gcc make -y
  • 下载rpm包

    wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-utils-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-static-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-common-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-devel-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/glibc-headers-2.17-55.el6.x86_64.rpm &wget http://copr-be.cloud.fedoraproject.org/results/mosquito/myrepo-el6/epel-6-x86_64/glibc-2.17-55.fc20/nscd-2.17-55.el6.x86_64.rpm &
  • 安装rpm包

    sudo rpm -Uvh *-2.17-55.el6.x86_64.rpm --force --nodeps
  • 安装完毕后再查看glibc版本,发现glibc版本已经到2.17了

    strings /lib64/libc.so.6 | grep GLIBC

3. 安装RabbitMQ

# 安装rpm -ivh socat-1.7.3.2-5.el7.lux.x86_64.rpm# 安装rpm -ivh rabbitmq-server-3.6.5-1.noarch.rpm

4. 开启管理界面及配置

# 开启管理界面rabbitmq-plugins enable rabbitmq_management# 修改默认配置信息 vim rabbitmq/lib/rabbitmq_server-3.6.5/ebin/rabbit.app # 比如修改密码、配置等等,例如:loopback_users 中的 <>,只保留guest

5. 启动

systemctl start rabbitmq-server# 启动服务systemctl stop rabbitmq-server# 停止服务systemctl restart rabbitmq-server# 重启服务systemctl status rabbitmq-server

设置配置文件

cd /usr/share/doc/rabbitmq-server-3.6.5/cp rabbitmq.config.example /etc/rabbitmq/rabbitmq.config

启动rabbitmq

cd /usr/lib/rabbitmq/lib/rabbitmq_server-3.6.5/ebin rabbitmq-server start &

常用命令

rabbitmq-server -deched--后台启动节点rabbitmqctl stop_app --关闭节点上的应用rabbitmqctl start_app --启动节点上的应用rabbitmqctl stop --关闭节点

修改了用户名之后,重启服务器造成创建用户丢失:

因为rabbitmq数据是根据当前hostname作为node节点作为数据名保存,这一点可以从rabbitmq-env文件中可以看出。

解决方案:

  1. 不要更改hostname
  2. https://blog.csdn.net/qq_27037443/article/details/78050526

6. 配置虚拟主机及用户

6.1 用户角色

RabbitMQ在安装好后,可以访问http://ip地址:15672 ;其自带了guest/guest的用户名和密码;如果需要创建自定义用户;那么也可以登录管理界面后,如下操作:

角色说明

1、 超级管理员(administrator)

可登陆管理控制台,可查看所有的信息,并且可以对用户,策略(policy)进行操作。

2、 监控者(monitoring)

可登陆管理控制台,同时可以查看rabbitmq节点的相关信息(进程数,内存使用情况,磁盘使用情况等)

3、 策略制定者(policymaker)

可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。

4、 普通管理者(management)

仅可登陆管理控制台,无法看到节点信息,也无法对策略进行管理。

5、 其他

无法登陆管理控制台,通常就是普通的生产者和消费者。

6.2 Virtual Hosts配置

像mysql拥有数据库的概念并且可以指定用户对库和表等操作的权限。RabbitMQ也有类似的权限管理;在RabbitMQ中可以虚拟消息服务器Virtual Host,每个Virtual Hosts相当于一个相对独立的RabbitMQ服务器,每个VirtualHost之间是相互隔离的。exchange、queue、message不能互通。 相当于mysql的db。Virtual Name一般以/开头。

创建Virtual Hosts

设置Virtual Hosts权限


RabbitMQ入门

1. 搭建示例工程

https://gitee.com/lixiaogou/rabbitmq

消息生产者com.itheima.producer.Producer_HelloWorld;

在执行上述的消息发送之后;可以登录rabbitMQ的管理控制台,可以发现队列和其消息:


消息消费者com.itheima.consumer.Consumer_HelloWorld;

2. 小结

上述的入门案例使用的是如下的简单模式:

在上图的模型中,有以下概念:

  • P:生产者,也就是要发送消息的程序
  • C:消费者:消息的接受者,会一直等待消息到来。
  • queue:消息队列,图中红色部分。可以缓存消息,生产者向其投递消息,消费者从其取出消息。

RabbitMQ工作模式

官网对应模式介绍:https://www.rabbitmq.com/getstarted.html

1. Helloworld简单模式

上面入门案例即是简单模式


2. Work queues工作队列模式

2.1 模式说明

Work Queues与入门程序的简单模式相比,多了一个或一些消费端,多个消费端竞争消费同一个队列中的消息。

应用场景:对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如发送短信服务

2.2 代码

Work Queues与入门程序的简单模式的代码是几乎一样的;可以完全复制,并复制多个消费者进行竞争消费测试。

生产者 Producer_WorkQueues

消费者1Consumer_WorkQueues1

消费者2Consumer_WorkQueues2

2.3 测试

启动两个消费者,然后再启动生产者发送消息;到IDEA的两个消费者对应的控制台查看是否竞争性的接收到消息。

2.4 小结

在一个Work queues队列中如果有多个消费者,那么消费者之间对于同一个消息的关系是竞争的关系。


3. Publish/Subscribe发布订阅模式☆

前面2个案例中,只有3个角色:

  • P:生产者,也就是要发送消息的程序
  • C:消费者:消息的接受者,会一直等待消息到来。
  • queue:消息队列,图中红色部分

3.1 模式说明

在订阅模型中,多了一个exchange角色,而且过程略有变化:

  • P:生产者,不再发送消息到队列中,而是发给X(交换机)
  • C:消费者,消息的接受者,会一直等待消息到来。
  • Queue:消息队列,接收消息、缓存消息。
  • Exchange:交换机,图中的X。一方面,接收生产者发送的消息。另一方面,知道如何处理消息,例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作,取决于Exchange的类型。Exchange有常见以下3种类型:
    • Fanout:广播,将消息交给所有绑定到交换机的队列
    • Direct:定向,把消息交给符合指定routing key 的队列
    • Topic:通配符,把消息交给符合routing pattern(路由模式) 的队列

Exchange(交换机)只负责转发消息,不具备存储消息的能力,因此如果没有任何队列与Exchange绑定,或者没有符合路由规则的队列,那么消息会丢失!

订阅模式示例图:

  1. 每个消费者监听自己的队列。
  2. 生产者将消息发给broker,由交换机将消息转发到绑定此交换机的每个队列,每个绑定交换机的队列都将接收到消息。

3.2. 代码

生产者Producer_PubSub

消费者1Consumer_PubSub1

消费者2Consumer_PubSub2

3.3 测试

启动所有消费者,然后使用生产者发送消息;在每个消费者对应的控制台可以查看到生产者发送的所有消息;到达广播的效果。

在执行完测试代码后,其实到RabbitMQ的管理后台找到Exchanges选项卡,点击 test_fanout 的交换机,可以查看到如下的绑定:

3.4 小结

交换机需要与队列进行绑定,绑定之后;一个消息可以被多个消费者收到。

发布订阅模式与工作队列模式的区别,重点还是在交换机上面

1、工作队列模式不用定义交换机,而发布/订阅模式需要定义交换机。

2、发布/订阅模式的生产方是面向交换机发送消息,工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。

3、发布/订阅模式需要设置队列和交换机的绑定,工作队列模式不需要设置,实际上工作队列模式会将队列绑定到默认的交换机 。


4. Routing路由模式

4.1 模式说明

路由模式特点:

  • 队列与交换机的绑定,不能是任意绑定了,而是要指定一个RoutingKey(路由key)
  • 消息的发送方在向Exchange发送消息时,也必须指定消息的 RoutingKey
  • Exchange不再把消息交给每一个绑定的队列,而是根据消息的Routing Key进行判断,只有队列的Routingkey与消息的 Routing key完全一致,才会接收到消息

图解:

  • P:生产者,向Exchange发送消息,发送消息时,会指定一个routing key。
  • X:Exchange(交换机),接收生产者的消息,然后把消息递交给 与routing key完全匹配的队列
  • C1:消费者,其所在队列指定了需要routing key 为 error 的消息
  • C2:消费者,其所在队列指定了需要routing key 为 info、error、warning 的消息

4.2 代码

在编码上与 Publish/Subscribe发布与订阅模式 的区别是交换机的类型为:Direct,还有队列绑定交换机的时候需要指定routing key。

生产者Producer_Routing

消费者1Consumer_Routing1

消费者2Consumer_Routing2

4.3 测试

启动所有消费者,然后使用生产者发送消息;消费者可以接收到生产者发送routing key对应队列的消息;达到按需接收的效果。

在执行完测试代码后,到RabbitMQ的管理后台找到Exchanges选项卡,点击 test_direct 交换机,可以查看到如下的绑定:

4.4 小结

Routing模式要求队列在绑定交换机时要指定routing key,消息会转发到符合routing key的队列。


5. Topics通配符模式

5.1. 模式说明

Topic类型与Direct相比,都可以根据RoutingKey把消息路由到不同的队列。但是Topic类型可以让队列在绑定Routing key 的时候使用通配符

Routingkey 一般是由一个或多个单词组成,多个单词之间以”.”分割,例如: item.insert

通配符规则:

#:匹配一个或多个词,多个占位符

*:只匹配一个词,一个占位符

举例:

item.#:能够匹配item.insert.abc 或者 item.insert

item.*:只能匹配item.insert

图解:

  • 红色Queue:绑定的是usa.# ,凡是以 usa.开头的routing key 都会被匹配到
  • 黄色Queue:绑定的是#.news ,凡是以 .news结尾的 routing key 都会被匹配

5.2. 代码

生产者Producer_Topics,所有error级别的日志存入数据库,所有order系统的日志存入数据库

消费者1Consumer_Topic1

消费者2Consumer_Topic2

5.3 测试

启动所有消费者,然后使用生产者发送消息;在消费者对应的控制台可以查看到生产者发送对应routing key对应队列的消息;到达按照需要接收的效果;并且这些routing key可以使用通配符。

在执行完测试代码后,其实到RabbitMQ的管理后台找到Exchanges选项卡,点击 test_topic 的交换机,可以查看到如下的绑定:

5.4. 小结

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 + Routing路由模式 的功能;

Topic在配置routing key 的时候可以使用通配符,更加灵活。


6. 工作模式总结

1、简单模式 HelloWorld

一个生产者、一个消费者,不需要设置交换机(使用默认的交换机)

2、工作队列模式 Work Queue

一个生产者、多个消费者(竞争关系),不需要设置交换机(使用默认的交换机)

3、发布订阅模式 Publish/subscribe

需要设置类型为fanout的交换机,并且交换机和队列进行绑定,当生产者发送消息到交换机后,交换机会将消息发送到绑定的队列,

4、路由模式 Routing

需要设置类型为direct的交换机,交换机和队列进行绑定,并且指定routing key,当生产者发送消息到交换机后,交换机会根据routing key将消息发送到对应的队列

5、通配符模式 Topic

需要设置类型为topic的交换机,交换机和队列进行绑定,并且指定通配符方式的routing key,当生产者发送消息到交换机后,交换机会根据routing key将消息发送到对应的队列


Spring 整合RabbitMQ

1. 搭建工程

https://gitee.com/lixiaogou/rabbitmq/tree/master/spring-rabbitmq-producers

https://gitee.com/lixiaogou/rabbitmq/tree/master/spring-rabbitmq-consumers

2. 消息监听器

队列监听器SpringQueueListener
广播监听器1 FanoutListener1
广播监听器2FanoutListener2
星号通配符监听器TopicListenerStar
井号通配符监听器TopicListenerWell
井号通配符监听器2TopicListenerWell2

3. 测试

ConsumerTest


Spring Boot整合RabbitMQ

1. 简介

在Spring项目中,可以使用Spring-Rabbit去操作RabbitMQ
https://github.com/spring-projects/spring-amqp

尤其是在spring boot项目中只需要引入对应的amqp启动器依赖即可,方便的使用RabbitTemplate发送消息,使用注解接收消息。

一般在开发过程中:

生产者工程:

  1. application.yml文件配置 R a b b i t M QRabbitMQRabbitMQ相关信息;

  2. 在生产者工程中编写配置类,用于创建交换机和队列,并进行绑定

  3. 注入RabbitTemplate对象,通过RabbitTemplate对象发送消息到交换机

消费者工程:

  1. application.yml文件配置 R a b b i t M QRabbitMQRabbitMQ相关信息

  2. 创建消息处理类,用于接收队列中的消息并进行处理

2. 搭建生产者工程

https://gitee.com/lixiaogou/rabbitmq/tree/master/springboot-rabbitmq-producer

2.1 创建工程

2.2 添加依赖

修改pom.xml文件内容为如下:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>com.itheima</groupId><artifactId>producer-springboot</artifactId><version>1.0-SNAPSHOT</version><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.4.RELEASE</version></parent><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId></dependency></dependencies></project>

2.3 启动类

package com.itheima;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;/** * @author lixiaolong */@SpringBootApplicationpublic class ProducerApplication {public static void main(String[] args) {SpringApplication.run(ProducerApplication.class);System.out.println("介就对了");}}

2.4 配置RabbitMQ

配置文件

创建application.yml,内容如下:

# 配置RabbitMQ的基本信息ip 端口 usernamepassword..spring:rabbitmq:host: localhost # ipport: 5672username: lxlpassword: lxlvirtual-host: /lxl
绑定交换机和队列

创建RabbitMQ队列与交换机绑定的配置类com.itheima.rabbitmq.config.RabbitMQConfig

package com.itheima.rabbitmq.config;import org.springframework.amqp.core.*;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;@Configurationpublic class RabbitMQConfig {public static final String EXCHANGE_NAME = "boot_topic_exchange";public static final String QUEUE_NAME = "boot_queue";/** * 1.交换机 * @return */@Beanpublic Exchange bootExchange() {return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();}/** * 2.Queue 队列 * @return */@Beanpublic Queue bootQueue() {return QueueBuilder.durable(QUEUE_NAME).build();}/** * 3. 队列和交互机绑定关系 Binding * @param queue队列 * @param exchange 交换机 * @return */@Beanpublic Binding bindQueueExchange(Queue queue, Exchange exchange) {return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();}}

3. 搭建消费者工程

https://gitee.com/lixiaogou/rabbitmq/tree/master/springboot-rabbitmq-consumer

3.1 创建工程

3.2 添加依赖

修改pom.xml文件内容为如下:

<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>2.1.4.RELEASE</version><relativePath/></parent><groupId>com.itheima</groupId><artifactId>consumer-springboot</artifactId><version>0.0.1-SNAPSHOT</version><properties><java.version>1.8</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

3.3 启动类

package com.itheima.rabbit.consumer;import org.springframework.boot.SpringApplication;import org.springframework.boot.autoconfigure.SpringBootApplication;@SpringBootApplicationpublic class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);System.out.println("介就对了");}}

3.4 配置RabbitMQ

创建application.yml,内容如下:

# 配置RabbitMQ的基本信息ip 端口 usernamepassword..spring:rabbitmq:host: localhost # ipport: 5672username: lxlpassword: lxlvirtual-host: /lxl

3.5 消息监听处理类

编写消息监听器com.itheima.rabbitmq.listener.RabbimtMQListener

package com.itheima.rabbit.consumer.listener;import org.springframework.amqp.core.Message;import org.springframework.amqp.rabbit.annotation.RabbitListener;import org.springframework.stereotype.Component;@Componentpublic class RabbimtMQListener {@RabbitListener(queues = "boot_queue")public void ListenerQueue(Message message) {System.out.println(new String(message.getBody()));}}

4. 测试

在生产者工程producer-springboot 中创建测试类,发送消息:

package com.itheima.test;import com.itheima.rabbitmq.config.RabbitMQConfig;import org.junit.Test;import org.junit.runner.RunWith;import org.springframework.amqp.rabbit.core.RabbitTemplate;import org.springframework.beans.factory.annotation.Autowired;import org.springframework.boot.test.context.SpringBootTest;import org.springframework.test.context.junit4.SpringRunner;@SpringBootTest@RunWith(SpringRunner.class)public class ProducerTest {//1.注入RabbitTemplate@Autowiredprivate RabbitTemplate rabbitTemplate;@Testpublic void testSend(){rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.haha","boot mq hello~~~");}}

先运行上述测试程序(交换机和队列才能先被声明和绑定),然后启动消费者;

在消费者工程consumer-springboot中控制台查看是否接收到对应消息。

另外,也可以在RabbitMQ的管理控制台中查看到交换机与队列的绑定: