一、MQ简介
1.1、什么是MQ消息队列
MQ全称 Message Queue（消息队列），是一种应用间的通信方式，消息发送后可以立即返回，由消息系统来确保消息的可靠传递。通过典型的 生产者和消费者模型,生产者不断向消息队列中生产消息，消费者不断的从队列中获取消息。因为消息的生产和消费都是异步的，而且只关心消息的发送和接收，这种异步处理的方式大大的节省了服务器的请求响应时间，从而提高了系统的吞吐量以及系统之间的解耦。利用高效可靠的消息传递机制进行平台无关的数据交流，并基于数据通信来进行分布式系统的集成。
1.2、MQ的优缺点
优点：
-
应用解耦：系统A要发送一个消息到多个系统，如果此时每增加一个系统或者删除一个系统，系统A都需要通过修改源码来增加接口，此时耦合非常高，但是如果中间使用消息队列的话，系统只需要发送一次到消息队列，别的系统就能复用该信息，当增加或删除系统调用接口的时候，不需要额外的更新代码。一个系统的耦合性越高，容错性就越低，可维护性就越低。
-
异步调用：还是ABCD四个系统，A系统收到一个请求，需要在自己本地写库，还需要往BCD三个系统写库，A系统自己写本地库需要3ms，往其他系统写库相对较慢，B系统200ms ，C系统350ms，D系统400ms，这样算起来，整个功能从请求到响应的时间为3ms+200ms+350ms+400ms=953ms，接近一秒，时间太长无法接受，如果用了MQ，用户发送请求到A系统耗时3ms，A系统发送三条消息到MQ，假如耗时5ms，用户从发送请求到相应3ms+5ms=8ms，仅用了8ms然后返回结果给用户，剩下的操作由BCD系统从Mq取出消息做自己的操作。
-
削峰填谷：削峰就是某一段时间用户请求突增时，由于系统不能处理这么高的并发量，就将多余的请求积压在MQ中，系统每次从MQ取出一定数量的请求比如1000，当处理完这1000个请求，再从MQ中取出1000请求。由于MQ在高峰期积压了很多的请求，在高峰期过后的一段时间内，消费消息的速度还是会维持在1000，直到消费完积压的消息，这就叫做“填谷”。
缺点：
- 系统可用性降低：系统引入的外部依赖越多，系统稳定性越差。一旦 MQ 宕机，就会对业务造成影响。需要保证MQ的高可用。
- 系统复杂度提高：MQ 的加入大大增加了系统的复杂度，以前系统间是同步的远程调用，现在是通过 MQ 进行异步调用。如何保证消息没有被重复消费？怎么处理消息丢失情况？怎么保证消息传递的顺序性？
- 一致性问题：A 系统处理完业务，通过 MQ 给B、C、D三个系统发消息数据，如果 B 系统、C 系统处理成功，D 系统处理失败。如何保证消息数据处理的一致性？
1.3、常见的MQ消息队列
| ActiveMQ | RabbitMQ | RocketMQ | kafka | |
|---|---|---|---|---|
| 公司/社区 | Apache | Rabbit | 阿里巴巴 | Apache |
| 协议支持 | OpenWire,STOMP，REST,XMPP,AMQP | AMQP，XMPP，SMTP，STOMP | 自定义 | 自定义协议，社区封装了http协议支持 |
| 客户端支持语言 | Java，C，C++，Python，PHP，Perl，.net等 | 官方支持Erlang，Java，Ruby等,社区产出多种API，几乎支持所有语言 | Java，C++ | 官方支持Java,社区产出多种API，如PHP，Python等 |
| 单机吞吐量 | 万级，吞吐量比RocketMQ和Kafka要低了一个数量级 | 万级，吞吐量比RocketMQ和Kafka要低了一个数量级 | 10万级，RocketMQ也是可以支撑高吞吐的一种MQ(最好) | 10万级别，这是kafka最大的优点，就是吞吐量高。一般配合大数据类的系统来进行实时数据计算、日志采集等场景 |
| topic数量对吞吐量的影响 | topic可以达到几百，几千个的级别，吞吐量会有较小幅度的下降这是RocketMQ的一大优势，在同等机器下，可以支撑大量的topic | topic从几十个到几百个的时候，吞吐量会大幅度下降所以在同等机器下，kafka尽量保证topic数量不要过多。如果要支撑大规模topic，需要增加更多的机器资源 | ||
| 时效性 | ms级 | 微秒级，这是rabbitmq的一大特点，延迟是最低的 | ms级 | 延迟在ms级以内 |
| 可用性 | 高，基于主从架构实现高可用性 | 高，基于主从架构实现高可用性 | 非常高，分布式架构 | 非常高，kafka是分布式的，一个数据多个副本，少数机器宕机，不会丢失数据，不会导致不可用 |
| 消息可靠性 | 有较低的概率丢失数据 | 经过参数优化配置，可以做到0丢失 | 经过参数优化配置，消息可以做到0丢失 | |
| 功能支持 | MQ领域的功能极其完备 | 基于erlang开发，所以并发能力很强，性能极其好，延时很低 | MQ功能较为完善，还是分布式的，扩展性好 | 功能较为简单，主要支持简单的MQ功能，在大数据领域的实时计算以及日志采集被大规模使用，是事实上的标准 |
| 优劣势总结 | 非常成熟，功能强大，在业内大量的公司以及项目中都有应用偶尔会有较低概率丢失消息而且现在社区以及国内应用都越来越少，官方社区现在对ActiveMQ 5.x维护越来越少几个月才发布一个版本而且确实主要是基于解耦和异步来用的，较少在大规模吞吐的场景中使用 | erlang语言开发，性能极其好，延时很低；吞吐量到万级，MQ功能比较完备而且开源提供的管理界面非常棒，用起来很好用社区相对比较活跃，几乎每个月都发布几个版本分在国内一些互联网公司近几年用rabbitmq也比较多一些但是问题也是显而易见的，RabbitMQ确实吞吐量会低一些，这是因为他做的实现机制比较重。而且erlang开发，国内有几个公司有实力做erlang源码级别的研究和定制？如果说你没这个实力的话，确实偶尔会有一些问题，你很难去看懂源码，你公司对这个东西的掌控很弱，基本职能依赖于开源社区的快速维护和修复bug。而且rabbitmq集群动态扩展会很麻烦，不过这个我觉得还好。其实主要是erlang语言本身带来的问题。很难读源码，很难定制和掌控。 | 接口简单易用，而且毕竟在阿里大规模应用过，有阿里品牌保障日处理消息上百亿之多，可以做到大规模吞吐，性能也非常好，分布式扩展也很方便，社区维护还可以，可靠性和可用性都是ok的，还可以支撑大规模的topic数量，支持复杂MQ业务场景而且一个很大的优势在于，阿里出品都是java系的，我们可以自己阅读源码，定制自己公司的MQ，可以掌控社区活跃度相对较为一般，不过也还可以，文档相对来说简单一些，然后接口这块不是按照标准JMS规范走的有些系统要迁移需要修改大量代码还有就是阿里出台的技术，你得做好这个技术万一被抛弃，社区黄掉的风险，那如果你们公司有技术实力我觉得用RocketMQ挺好的 | kafka的特点其实很明显，就是仅仅提供较少的核心功能，但是提供超高的吞吐量，ms级的延迟，极高的可用性以及可靠性，而且分布式可以任意扩展同时kafka最好是支撑较少的topic数量即可，保证其超高吞吐量而且kafka唯一的一点劣势是有可能消息重复消费，那么对数据准确性会造成极其轻微的影响，在大数据领域中以及日志采集中，这点轻微影响可以忽略这个特性天然适合大数据实时计算以及日志收集 |
1.4、RabbitMQ的背景
RabbitMQ是一个由erlang开发的AMQP（Advanced Message Queue ）的开源实现。实现MQ大致有两种主流方式：AMQP、JMS。
- AMQP(Advanced Message Queue)：高级消息队列协议，一个提供统消息服务的应用层标准高级消息队列协议。它是应用层协议的一个开放标准，为面向消息的中间件设计，基于此协议的客户端与消息中间件可传递消息，并不受产品、开发语言等条件的限制。AMQP是一个二进制协议，拥有一些现代化特点：多信道、协商式，异步，安全，扩平台，中立，高效。
- JMS(JavaMessage Service):Java消息服务,应用程序接口，是一个Java平台中关于面向消息中间件（MOM）的API，用于在两个应用程序之间，或分布式系统中发送消息，进行异步通信。
AMQP 与 JMS 区别:
- JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
- JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
- JMS规定了两种消息模式；而AMQP的消息模式更加丰富
简单来说AMQP是一组协议，而JMS是一套Java的API
详细了解区别可以参考这篇博客:https://blog.csdn.net/hpttlook/article/details/23391967
二、RabbitMQ的基本概念
Rabbit的基础架构如下：
-
Broker：表示消息队列服务器实体，接收和分发消息的应用。
-
Virtual host：当多个不同的用户使用同一个 RabbitMQ server 提供的服务时，可以划分出多个vhost，每个用户在自己的 vhost 创建 exchange／queue 等。每个 vhost 本质上就是一个 mini 版的 RabbitMQ 服务器，拥有自己的队列、交换器、绑定和权限机制。vhost 是 AMQP 概念的基础，必须在连接时指定，RabbitMQ 默认的 vhost 是 / 。
-
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) and fanout (multicast)
-
Queue：消息队列，用来保存消息直到被消费者取走。它是消息的容器，也是消息的终点。一个消息可投入一个或多个队列。消息一直在队列里面，等待消费者连接到这个队列将其取走。
-
Binding：exchange 和 queue 之间的虚拟连接，binding 中可以包含 routing key。Binding 信息被保存到 exchange 中的查询表中，用于 message 的分发依据
-
Publisher：消息的生产者，也是一个向交换器发布消息的客户端应用程序。
-
Consumer：消息的消费者，表示一个从消息队列中取得消息的客户端应用程序。
-
Message：消息，消息是不具名的，它由消息头和消息体组成。消息体是不透明的，而消息头则由一系列的可选属性组成，这些属性包括routing-key（路由键）、priority（相对于其他消息的优先权）、delivery-mode（指出该消息可能需要持久性存储）等。
RabbitMQ官方地址：http://www.rabbitmq.com/
三、RabbitMQ的安装及配置
- 下载RabbitMq和Erlang的依赖包上传到Centos7
下载地址：https://www.rabbitmq.com/download.html
RabbitMq是由Erlang编写的在安装RabbitMQ之前需要先安装Erlang
-
安装
# 1.安装Erlang依赖包 rpm -ivh erlang-22.0.7-1.el7.x86_64.rpm # 2.安装RabbitMQ安装包(需要联网) #默认安装完成后rabbitmq有个配置文件模板/usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example，需要将配置文件复制到/etc/rabbitmq/目录中,并修改名称为rabbitmq.config yum install -y rabbitmq-server-3.7.18-1.el7.noarch.rpm # 3.复制配置文件到/etc/rabbitmq/目录下并将名字改为rabbitmq.config cp /usr/share/doc/rabbitmq-server-3.7.18/rabbitmq.config.example /etc/rabbitmq/rabbitmq.config # 4.修改配置文件 打开guest用户 vim /etc/rabbitmq/rabbitmq.config
将{loopback_users, []},前边的‘%%’注释和后边的逗号删掉
# 5.rabbitmq有一个图形化的管理界面需要启动这个插件，需要从浏览器打开这个界面默认端口为15672 rabbitmq-plugins enable rabbitmq_management # 6.启动RabbitMQ服务 systemctl start rabbitmq-server # 7. 关闭防火墙 systemctl disable firewalld
可以通过http://IP地址:15672访问web管理界面，账户为guest，密码也为guest
注意：如果使用的云服务器需要在控制台安全组放行端口
-
通过UI界面添加用户和虚拟主机
- 创建用户
上面的Tags选项，其实是指定用户的角色，可选的有以下几个：
-
超级管理员(administrator)可登陆管理控制台，可查看所有的信息，并且可以对用户，策略(policy)进行操作。
-
监控者(monitoring)可登陆管理控制台，同时可以查看rabbitmq节点的相关信息(进程数，内存使用情况，磁盘使用情况等)
-
策略制定者(policymaker)可登陆管理控制台, 同时可以对policy进行管理。但无法查看节点的相关信息(上图红框标识的部分)。
-
普通管理者(management)仅可登陆管理控制台，无法看到节点信息，也无法对策略进行管理。
-
其他无法登陆管理控制台，通常就是普通的生产者和消费者。
-
创建虚拟主机
-
绑定用户和虚拟主机
-
RabbitMq管理行命令
# RabbitMQ除了有web段的图形化管理页面也可以使用命令
# 查看RabbitMQ的命令
rabbitmqctl help
# 查看RabbitMQ版本
rabbitmqctl version
# 查看RabbitMQ插件的命令
rabbitmq-plugins help
# 查看插件是否启动
rabbitmq_management is_enabled
#查看插件
rabbitmq-plugins list
rabbitmqctl help 展现的部分命令截图：
rabbitmq-plugins help 命令截图：
安装RabbitMq8可一看看这篇文章:https://blog.csdn.net/weixin_40584261/article/details/106826044
四、RabbitMQ的工作模式
在 RabbitMQ 官网上提供了 6 中工作模式：简单模式、工作队列模式、发布/订阅模式、路由模式、主题模式 和 RPC 模式以及最新的发布确认模式。
1、Hello Word简单模式
简单工作模式由 3 个对象组成：生产者、队列、消费者。只有一个消费者，生产者将消息放入队列，消费者监听消息队列，如果队列中有消息,就消费掉,消息被拿走后,自动从队列中删除。
-
创建Maven工程导入RabbitMQ的java客户端依赖
<dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.7.2</version> </dependency> -
获取与RabbitMQ连接的工具类
public class RabbitMqUtil { //连接工厂，用来创建连接 private static ConnectionFactory connectionFactory; static { connectionFactory = new ConnectionFactory(); //主机地址 connectionFactory.setHost("48.122.183.82"); //rabbitMq默认端口号是5672，UI管理界面是15672 connectionFactory.setPort(5672); //设置虚拟主机，rabbitmq默认虚拟机名称为“/”，虚拟机相当于一个独立的mq服务虚拟机相当于一个独立的mq服务，可以自己在UI界面新建一个虚拟主机并添加用户 connectionFactory.setVirtualHost("/rabbitmqdemo"); //账户 connectionFactory.setUsername("admin"); //密码 connectionFactory.setPassword("admin"); } public static Connection getConnection() { try { //创建与RabbitMQ服务的TCP连接 return connectionFactory.newConnection(); } catch (IOException e) { e.printStackTrace(); } catch (TimeoutException e) { e.printStackTrace(); } return null; } public static void closeConnectionAndChannel(Connection connection, Channel channel) { try { if (channel != null) { //关闭通道 channel.close(); } } catch (TimeoutException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } try { if (connection != null) { //关闭连接 connection.close(); } } catch (IOException e) { e.printStackTrace(); } } }3.生产者
public class Provider { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = RabbitMqUtil2.getConnection(); //创建与Exchange的通道，每个连接可以创建多个通道 Channel channel = connection.createChannel(); //通道绑定消息队列,如果队列不存在则自动创建 //参数1：队列的名字 // 参数2用来定义队列是否持久化 true为持久化(rabbit服务关闭就会丢失队列，队列中的消息也会丢失，设置为true队列可持久化，要想消息也被持久化需要额外设置) // 参数3exclusive：是否独占队列(只允许当前连接可用) //参数4：是否在消费完成消费后(消费者与队列断开连接)自动删除队列 //参数5 ： 额外的附加参数 //注意：生产者与消费者队列的参数必须一致!!!!! channel.queueDeclare("hello",true,false,false,null); //这里没有指定交换机，消息将发送给默认交换机，每个队列也会绑定那个默认的交换机，但是不能显 示绑定或解除绑定,默认的交换机，routingKey等于队列名称 //发布消息 //参数1： 交换机名称 //参数2：参数队列名称 //参数3：传递消息的额外设置 MessageProperties.PERSISTENT_TEXT_PLAIN :消息持久化 //参数4：消息的具体内容 channel.basicPublish("","hello", MessageProperties.PERSISTENT_TEXT_PLAIN,("hello rabbitmq"+new Date()).getBytes()); RabbitMqUtil.closeConnectionAndChannel(connection,channel); } }4.消费者
/** *消费者开启后会一直消费队列中的消息 * hello word模型 只有一个消费者 一对一 交换机参数为空字符串表示默认交换机，生产者生产消息后直接**放入消息队列 */ public class Consumer { public static void main(String[] args) throws IOException, TimeoutException { Connection connection = RabbitMqUtil.getConnection(); Channel channel = connection.createChannel(); //声明队列 channel.queueDeclare("hello", false,false,false,null); //参数1：消费哪个队列的消息 //参数2：开启消息的自动确认机制 //参数3：消费消息时的回调接口 channel.basicConsume("hello",true,new DefaultConsumer(channel){ //body 消息队列中取出的消息 @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { System.out.println(new String(body)); System.out.println("consumerTag = " + consumerTag); System.out.println("DeliveryTag = " +envelope.getDeliveryTag()); System.out.println("Redeliver = "+envelope.isRedeliver()); } }); } }
2、work queue工作队列模式
Work Queues与入门程序的简单模式相比，多了一个或一些消费端，多个消费端共同消费同一个队列中的消息。默认情况下，RabbitMQ将按顺序将每个消息发送给下一个使用者。平均而言，每个消费者都会收到相同数量的消息。这种分发消息的方式称为循环。对于 任务过重或任务较多情况使用工作队列可以提高任务处理的速度。
1.生产者
public class Provider {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
//通过通道绑定队列
channel.queueDeclare("workqueue",true,false,false,null);
for (int i = 0; i < 30; i++) {
channel.basicPublish("","workqueue", MessageProperties.PERSISTENT_BASIC,("workqueue消息"+i).getBytes());
}
RabbitMqUtil.closeConnectionAndChannel(connection,channel);
}
}
/**
可以多写几个消费者
*/
class Consumer1 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
//通道绑定队列
channel.queueDeclare("workqueue",true,false,false,null);
//每次分配一个消息，确认一个消息，才又分配一个消息,为了使消费者做到能者多劳
channel.basicQos(1);
//处理消息第二个参数为false表示不自动确认
channel.basicConsume("workqueue",false,new DefaultConsumer(channel){
/**
* consumerTag 消息者标签，在channel.basicConsume时候可以指定
* envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
* properties 属性信息
* body 消息
*/
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
e.printStackTrace();
}
//路由key
System.out.println("路由key为：" + envelope.getRoutingKey());
//交换机
System.out.println("交换机为：" + envelope.getExchange());
//消息id
System.out.println("消息id为：" + envelope.getDeliveryTag());
System.out.println("处理消息"+new String(body));
//手动确认消息已经被处理
//参数1：消息标识
//参数2： 是否同时确认多个消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}
}
class Consumer2 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
//通道绑定队列
channel.queueDeclare("workqueue",true,false,false,null);
//每次分配一个消息，确认一个消息，才又分配一个消息
channel.basicQos(1);
//处理消息
channel.basicConsume("workqueue",false,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
try {
TimeUnit.SECONDS.sleep(2);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("处理消息"+new String(body));
//手动确认消息已经被处理
//参数1：消息标识
//参数2： 是否同时确认多个消息
channel.basicAck(envelope.getDeliveryTag(),false);
}
});
}
}
3、Publish/Subscribe发布与订阅模式
Exchange分发消息时根据类型的不同分发策略有区别，目前共四种类型：direct、fanout、topic、headers 。headers 匹配 AMQP 消息的 header 而不是路由键，此外 headers 交换器和 direct 交换器完全一致，但性能差很多，目前几乎用不到了。发布订阅模式，交换机类型设置为fanout。
- 、每个消费者监听自己的队列。
- 生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收到消息。fanout 类型转发消息是最快的。
生产者
public class Provider {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
//将通道声明指定交换机
//参数1：交换机名称
//参数2: 交换机类型 fanout 广播类型
channel.exchangeDeclare("logs","fanout");
//发送消息,fanout交换机这里的路由key在广播下几乎没什么作用
channel.basicPublish("logs","",null,"fanout message queue".getBytes());
RabbitMqUtil.closeConnectionAndChannel(connection,channel);
}
}
消费者
/**
可以多写几个消费者
*/
class Consumer1 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
//通道绑定交换机
channel.exchangeDeclare("logs","fanout");
//临时队列
String queue = channel.queueDeclare().getQueue();
//交换机绑定临时队列
//参数1：队列名称
//参数2：交换机名称
//参数三：routekey，这里用不到
channel.queueBind(queue,"logs","");
//消费消息
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1："+ new String(body));
}
});
}
}
class Consumer2 {
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
//通道绑定交换机
channel.exchangeDeclare("logs","fanout");
//临时队列
String queue = channel.queueDeclare().getQueue();
//交换机绑定临时队列
channel.queueBind(queue,"logs","");
//消费消息
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2："+ new String(body));
}
});
}
}
交换机需要与队列进行绑定，绑定之后；一个消息可以被多个消费者都收到。
发布订阅模式与工作队列模式的区别：
1、工作队列模式不用定义交换机，而发布/订阅模式需要定义交换机。
2、发布/订阅模式的生产方是面向交换机发送消息，工作队列模式的生产方是面向队列发送消息(底层使用默认交换机)。
3、发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑 定到默认的交换机 。
4、 Routing路由模式
交换机的类型设置为direct
- 队列与交换机的绑定，不能是任意绑定了，而是要指定一个
RoutingKey（路由key）,队列可以指定多个RoutingKey; - 消息的发送方在 向 Exchange发送消息时，也必须指定消息的
RoutingKey。 - Exchange不再把消息交给每一个绑定的队列，而是根据消息的
Routing Key进行判断，只有队列的Routingkey与消息的Routing key完全一致，才会接收到消息
生产者
public class Provider {
public static void main(String[] args) throws IOException {
//连接
Connection connection = RabbitMqUtil.getConnection();
//通道
Channel channel = connection.createChannel();
//交换机，参数1：交换机名称，参数2：类型路由模式
channel.exchangeDeclare("logs_direct","direct");
//路由key
String routeKey = "error";
//发送消息
channel.basicPublish("logs_direct",routeKey,null,("路由消息key="+routeKey).getBytes());
RabbitMqUtil.closeConnectionAndChannel(connection,channel);
}
}
消费者
class Consumer1 {
public static void main(String[] args) throws IOException {
//连接
Connection connection = RabbitMqUtil.getConnection();
//通道
Channel channel = connection.createChannel();
//交换机
channel.exchangeDeclare("logs_direct","direct");
//创建临时队列
String queue = channel.queueDeclare().getQueue();
//基于route key 绑定队列和交换机
channel.queueBind(queue,"logs_direct","error");
channel.basicConsume(queue,true, new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1："+new String(body));
}
});
}
}
class Comsumer2{
public static void main(String[] args) throws IOException {
Connection connection = RabbitMqUtil.getConnection();
Channel channel = connection.createChannel();
channel.exchangeDeclare("logs_direct","direct");
String queue = channel.queueDeclare().getQueue();
channel.queueBind(queue,"logs_direct","info");
channel.queueBind(queue,"logs_direct","error");
channel.queueBind(queue,"logs_direct","warn");
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2："+ new String(body));
}
});
}
}
5.Topics通配符模式
Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定Routing key 的时候使用通配符！
Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割.
：匹配0个或者多个单词
*：配置一个单词
生产者
public class Provider {
public static void main(String[] args) throws IOException {
//连接
Connection connection = RabbitMqUtil.getConnection();
//通道
Channel channel = connection.createChannel();
//交换机
channel.exchangeDeclare("ex_topic","topic");
//路由key
String routeKey = "topic";
//生产
channel.basicPublish("ex_topic",routeKey,null,"topic消息队列".getBytes());
//关闭资源
RabbitMqUtil.closeConnectionAndChannel(connection,channel);
}
}
消费者
class Consumer1 {
public static void main(String[] args) throws IOException {
//连接
Connection connection = RabbitMqUtil.getConnection();
//通道
Channel channel = connection.createChannel();
//交换机
channel.exchangeDeclare("ex_topic","topic");
//消息队列
String queue = channel.queueDeclare().getQueue();
//交换机绑定消息队列
channel.queueBind(queue,"ex_topic","*.topic");
//消费
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者1："+new String(body));
}
});
}
}
class Consumer2 {
public static void main(String[] args) throws IOException {
//连接
Connection connection = RabbitMqUtil.getConnection();
//通道
Channel channel = connection.createChannel();
//交换机
channel.exchangeDeclare("ex_topic","topic");
//消息队列
String queue = channel.queueDeclare().getQueue();
//交换机绑定消息队列
channel.queueBind(queue,"ex_topic","topic");
//消费
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者2："+new String(body));
}
});
}
}
class Consumer3 {
public static void main(String[] args) throws IOException {
//连接
Connection connection = RabbitMqUtil.getConnection();
//通道
Channel channel = connection.createChannel();
//交换机
channel.exchangeDeclare("ex_topic","topic");
//消息队列
String queue = channel.queueDeclare().getQueue();
//交换机绑定消息队列
channel.queueBind(queue,"ex_topic","topic.#");
//消费
channel.basicConsume(queue,true,new DefaultConsumer(channel){
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("消费者3："+new String(body));
}
});
}
}
五、Spring整合RabbitMQ
生产者：
添加依赖:
<dependencies>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<version>5.1.7.RELEASE</version>
</dependency>
<dependency>
<groupId>org.springframework.amqp</groupId>
<artifactId>spring-rabbit</artifactId>
<version>2.1.8.RELEASE</version>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>4.12</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-test</artifactId>
<version>5.1.7.RELEASE</version>
</dependency>
</dependencies>
创建spring-rabbitmq-producer.xml配置文件
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="74.121.83.182"
port="5672"
username="guest"
password="guest"
virtual-host="/rabbitmqdemo"/>
<!--定义管理交换机、队列-->
<rabbit:admin connection-factory="connectionFactory"/>
<!--定义持久化队列，不存在则自动创建；不绑定到交换机则绑定到默认交换机
默认交换机类型为direct，名字为：""，路由键为队列的名称
-->
<!--
id：bean的名称
name：queue的名称
auto-declare:自动创建
auto-delete:自动删除。 最后一个消费者和该队列断开连接后，自动删除队列
exclusive:是否独占
durable：是否持久化
-->
<rabbit:queue id="spring_queue" name="spring_queue" auto-declare="true"/>
<!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~广播；所有队列都能收到消息~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->
<!--定义广播交换机中的持久化队列，不存在则自动创建-->
<rabbit:queue id="spring_fanout_queue_1" name="spring_fanout_queue_1" auto-declare="true" />
<!--定义广播交换机中的持久化队列，不存在则自动创建-->
<rabbit:queue id="spring_fanout_queue_2" name="spring_fanout_queue_2" auto-declare="true"/>
<!--定义fanout类型交换机；并绑定上述两个队列-->
<rabbit:fanout-exchange id="spring_fanout_exchange" name="spring_fanout_exchange" auto-declare="true">
<rabbit:bindings>
<rabbit:binding queue="spring_fanout_queue_1" />
<rabbit:binding queue="spring_fanout_queue_2"/>
</rabbit:bindings>
</rabbit:fanout-exchange>
<!--direct类型的交换机-->
<!-- <rabbit:direct-exchange name="aa" >
<rabbit:bindings>
<!–direct 类型的交换机绑定队列 key ：路由key queue：队列名称–>
<rabbit:binding queue="spring_queue" key="xxx"></rabbit:binding>
</rabbit:bindings>
</rabbit:direct-exchange>-->
<!-- ~~~~~~~~~~~~~~~~~~~~~~~~~~~~通配符；*匹配一个单词，#匹配多个单词 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~ -->
<!--定义广播交换机中的持久化队列，不存在则自动创建-->
<rabbit:queue id="spring_topic_queue_star" name="spring_topic_queue_star" auto-declare="true"/>
<!--定义广播交换机中的持久化队列，不存在则自动创建-->
<rabbit:queue id="spring_topic_queue_well" name="spring_topic_queue_well" auto-declare="true"/>
<!--定义广播交换机中的持久化队列，不存在则自动创建-->
<rabbit:queue id="spring_topic_queue_well2" name="spring_topic_queue_well2" auto-declare="true"/>
<rabbit:topic-exchange id="spring_topic_exchange" name="spring_topic_exchange" auto-declare="true">
<rabbit:bindings>
<rabbit:binding pattern="topic.*" queue="spring_topic_queue_star"/>
<rabbit:binding pattern="topic.#" queue="spring_topic_queue_well"/>
<rabbit:binding pattern="*.topic.#" queue="spring_topic_queue_well2"/>
</rabbit:bindings>
</rabbit:topic-exchange>
<!--定义rabbitTemplate对象操作可以在代码中方便发送消息-->
<rabbit:template id="rabbitTemplate" connection-factory="connectionFactory"/>
</beans>
在test目录下创建测试类
发送消息
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = "classpath:spring-rabbitmq-producer.xml")
public class ProducerTest {
@Autowired
private RabbitTemplate rabbitTemplate;
/**
* 只发队列消息
* 默认交换机类型为 direct
* 交换机的名称为空，路由键为队列的名称
*/
@Test
public void queueTest(){
//路由键与队列同名
rabbitTemplate.convertAndSend("spring_queue", "只发队列spring_queue的消息。");
}
/**
* 发送广播
* 交换机类型为 fanout
* 绑定到该交换机的所有队列都能够收到消息
*/
@Test
public void fanoutTest(){
/**
* 参数1：交换机名称
* 参数2：路由键名（广播设置为空）
* 参数3：发送的消息内容
*/
rabbitTemplate.convertAndSend("spring_fanout_exchange", "", "发送到spring_fanout_exchange交换机的广播消息");
}
/**
* 通配符
* 交换机类型为 topic
* 匹配路由键的通配符，*表示一个单词，#表示多个单词
* 绑定到该交换机的匹配队列能够收到对应消息
*/
@Test
public void topicTest(){
/**
* 参数1：交换机名称
* 参数2：路由键名
* 参数3：发送的消息内容
*/
rabbitTemplate.convertAndSend("spring_topic_exchange", "topic", "发送到spring_topic_exchange交换机htopic的消息");
}
}
消费者：
导入pom依赖同生产者一样
创建各个监听器类
public class SpringTopicListener implements MessageListener {
@Override
public void onMessage(Message message) {
System.out.println(new String(message.getBody()));
}
}
public class SpringQueueListener implements MessageListener {
@Override
public void onMessage(Message message) {
//打印消息
System.out.println(new String(message.getBody()));
}
}
编写spring-rabbitmq-consumer.xml
<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xmlns:context="http://www.springframework.org/schema/context"
xmlns:rabbit="http://www.springframework.org/schema/rabbit"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/context
https://www.springframework.org/schema/context/spring-context.xsd
http://www.springframework.org/schema/rabbit
http://www.springframework.org/schema/rabbit/spring-rabbit.xsd">
<!-- 定义rabbitmq connectionFactory -->
<rabbit:connection-factory id="connectionFactory" host="74.121.83.182"
port="5672"
username="guest"
password="guest"
virtual-host="/rabbitmqdemo"/>
<rabbit:admin connection-factory="connectionFactory"/>
<!-- 配置两个监听器-->
<bean id="springQueueListener" class="com.rabbitmq.listener.SpringQueueListener"/>
<bean id="springTopicListener" class="com.rabbitmq.listener.SpringTopicListener"/>
<rabbit:listener-container connection-factory="connectionFactory" auto-declare="true">
<rabbit:listener ref="springQueueListener" queue-names="spring_queue"/>
<rabbit:listener ref = "springTopicListener" queue-names="spring_topic_queue_well"/>
</rabbit:listener-container>
</beans>
六、SpringBoot整合RabbitMQ
6.1、注解的方式
创建SpringBoot工程引入mq的starter
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
编写yml配置文件
spring:
application:
name: springboot_rabbitmq
rabbitmq:
host: 74.121.83.182
port: 5672
username: guest
password: guest
virtual-host: /rabbitmqdemo
-
简单hello word模型
生产者：
@Autowired private RabbitTemplate rabbitTemplate; @Test public void testHello(){ /** *hello 为队列名称 */ rabbitTemplate.convertAndSend("hello","hello world"); }消费者：
@Component @RabbitListener(queuesToDeclare = @Queue("hello")) public class HelloCustomer { @RabbitHandler public void receive1(String message){ System.out.println("message = " + message); } } -
work queue工作队列模型
生产者：
@Autowired private RabbitTemplate rabbitTemplate; @Test public void testWork(){ for (int i = 0; i < 10; i++) { /** *work为队列名称 */ rabbitTemplate.convertAndSend("work","hello work!"); } }消费者：
@Component public class WorkCustomer { /** *@RabbitListenerke可以直接注解在方法上，就不需要@RabbitHandler */ @RabbitListener(queuesToDeclare = @Queue("work")) public void receive1(String message){ System.out.println("work message1 = " + message); } @RabbitListener(queuesToDeclare = @Queue("work")) public void receive2(String message){ System.out.println("work message2 = " + message); } }:默认在Spring AMQP实现中Work这种方式就是公平调度,如果需要实现能者多劳需要在yaml中配置
spring: rabbitmq: listener: simple: prefetch: 1 -
fanout广播模型
生产者
@Autowired private RabbitTemplate rabbitTemplate; @Test public void testFanout() throws InterruptedException { rabbitTemplate.convertAndSend("logs","","这是日志广播"); }消费者
@Component public class FanoutCustomer { @RabbitListener(bindings = @QueueBinding( value = @Queue,////创建临时队列 exchange = @Exchange(name="logs",type = "fanout") )) public void receive1(String message){ System.out.println("message1 = " + message); } @RabbitListener(bindings = @QueueBinding( value = @Queue, //创建临时队列 exchange = @Exchange(name="logs",type = "fanout") //绑定交换机类型 )) public void receive2(String message){ System.out.println("message2 = " + message); } } -
Route路由模型
生产者
@Autowired private RabbitTemplate rabbitTemplate; @Test public void testDirect(){ rabbitTemplate.convertAndSend("directs","error","error 的日志信息"); }消费者
@Component public class DirectCustomer { @RabbitListener(bindings ={ @QueueBinding( value = @Queue(),//没有指定名字就是零食队列 key={"info","error"},//route_key exchange = @Exchange(type = "direct",name="directs") )}) public void receive1(String message){ System.out.println("message1 = " + message); } @RabbitListener(bindings ={ @QueueBinding( value = @Queue(), key={"error"}, exchange = @Exchange(type = "direct",name="directs") )}) public void receive2(String message){ System.out.println("message2 = " + message); } } -
Topic 订阅模型
生产者
@Autowired private RabbitTemplate rabbitTemplate; //topic @Test public void testTopic(){ rabbitTemplate.convertAndSend("topics","user.save.findAll","user.save.findAll 的消息"); }消费者
@Component public class TopCustomer { @RabbitListener(bindings = { @QueueBinding( value = @Queue, key = {"user.*"}, exchange = @Exchange(type = "topic",name = "topics") ) }) public void receive1(String message){ System.out.println("message1 = " + message); } @RabbitListener(bindings = { @QueueBinding( value = @Queue, key = {"user.#"}, exchange = @Exchange(type = "topic",name = "topics") ) }) public void receive2(String message){ System.out.println("message2 = " + message); } }
配置类的方式：
编写配置类：
@Configuration
public class RabbitMQConfig {
/**
* 交换机
* @return
*/
@Bean("bootExchange")
public Exchange bootExchange(){
return ExchangeBuilder.topicExchange("boot_topic").durable(true).autoDelete().build();
}
/**
* 队列
* @return
*/
@Bean("bootQueue")
public Queue bootQueue(){
return QueueBuilder.durable("boot_queue").build();
}
/**
* 绑定交换机和队列
* @param queue
* @param exchange
* @return
*/
@Bean
public Binding bindQueueExchange(@Qualifier("bootQueue") Queue queue,@Qualifier("bootExchange") Exchange exchange){
return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
}
}
生产者:
@Test
void contextLoads() {
rabbitTemplate.convertAndSend("boot_topic","boot.test","hello ！ SpringBoot整合RabbitMQ");
}
消费者：
@RabbitListener(queues = "boot_queue")
public void rabbitmqListener(Message message){
System.out.println(message);
}
以上是个人笔记，有错误请大家指出