本文最后更新于 1228 天前，其中的信息可能已经有所发展或是发生改变。

RabbitMQ学习札记

1.MQ概述

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

2.MQ 的优势和劣势

优势：

应用解耦：提高系统容错性和可维护性
异步提速：提升用户体验和系统吞吐量
削峰填谷：提高系统稳定性

如订单系统，在下单的时候就会往数据库写数据。但是数据库只能支撑每秒1000左右的并发写入，并发量再高就容易宕机。低峰期的时候并发也就100多个，但是在高峰期时候，并发量会突然激增到5000以上，这个时候数据库肯定卡死了。

消息被MQ保存起来了，然后系统就可以按照自己的消费能力来消费，比如每秒1000个数据，这样慢慢写入数据库，这样就不会卡死数据库了。

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

劣势：

系统可用性降低
系统复杂度提高
一致性问题

3.AMQP 和 JMS

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

3.1. AMQP

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

3.2. JMS

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

3.3. AMQP 与 JMS 区别

JMS是定义了统一的接口，来对消息操作进行统一；AMQP是通过规定协议来统一数据交互的格式
JMS限定了必须使用Java语言；AMQP只是协议，不规定实现方式，因此是跨语言的。
JMS规定了两种消息模式；而AMQP的消息模式更加丰富

4.常见的 MQ 产品

目前业界有很多的 MQ 产品，例如 RabbitMQ、RocketMQ、ActiveMQ、Kafka、ZeroMQ、MetaMq等，也有直接使用 Redis 充当消息队列的案例。

市场上常见的消息队列有如下：

ActiveMQ：基于JMS
RabbitMQ：基于AMQP协议，erlang语言开发，稳定性好
RocketMQ：基于JMS，阿里巴巴产品
Kafka：类似MQ的产品；分布式消息系统，高吞吐量

5.RabbitMQ 简介

RabbitMQ是由erlang语言开发，基于AMQP（Advanced Message Queue 高级消息队列协议）协议实现的消息队列，它是一种应用程序之间的通信方法，消息队列在分布式系统开发中应用非常广泛。

RabbitMQ官方地址：http://www.rabbitmq.com/

RabbitMQ提供了6种模式：简单模式，work模式，Publish/Subscribe发布与订阅模式，Routing路由模式，Topics主题模式，RPC远程调用模式~~（远程调用，不太算MQ；暂不作介绍~~）；

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

6.安装及配置RabbitMQ

6.1 安装依赖环境

安装依赖环境：

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

根据官方建议要先安装：

yum install socat

yum install logrotate

再安装erlang

下载：https://packages.erlang-solutions.com/erlang/rpm/centos/7/x86_64/esl-erlang_22.3.2-1~centos~7_amd64.rpm

安装：rpm -Uvh esl-erlang_22.3.2-1_centos_7_amd64.rpm --force --nodeps

最后安装rabbitmq

这个页面是rabbitmq对应的erlang版本查询网站：https://www.rabbitmq.com/which-erlang.html

下载：https://github.com/rabbitmq/rabbitmq-server/releases/download/v3.8.3/rabbitmq-server-3.8.3-1.el7.noarch.rpm

安装：rpm -ivh rabbitmq-server-3.8.3-1.el7.noarch.rpm

安装路径默认在：

/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.3/ebin

6.2开启管理界面及配置

# 开启rabbitmq management及远程登录
rabbitmq-plugins enable rabbitmq_management

# 设置guest远程访问：
vi /usr/lib/rabbitmq/lib/rabbitmq_server-3.8.3/ebin/rabbit.app
# loopback_users里的<<"guest">>删除，重启rabbitmq服务
# 比如修改密码、配置等等，例如：loopback_users 中的 <<"guest">>,只保留guest

要访问RabbitMQ的管理面板，请使用您最喜爱的Web浏览器并打开以下URL：

http://Your_Server_IP:15672/

6.3 启动停止服务

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

6.4 设置配置文件

如果不知道配置文件在哪，可以查找：

[root@wydsj-gpapp04 ~]# find / -name rabbitmq-defaults
/usr/lib/rabbitmq/bin/rabbitmq-defaults
/usr/lib/rabbitmq/lib/rabbitmq_server-3.8.3/sbin/rabbitmq-defaults

6.5 配置虚拟主机及用户

角色说明：

1、超级管理员(administrator)

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

2、监控者(monitoring)

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

3、策略制定者(policymaker)

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

4、普通管理者(management)

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

5、其他

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

配置有两种方式：

命令行
界面

6.5.1 命令行方式：

创建用户：

创建用户名hcxtadmin，密码hcxtadmin123 的用户

rabbitmqctl add_user hcxtadmin hcxtadmin123

设置超级管理员

设置hcxtadmin为超级管理员

rabbitmqctl set_user_tags hcxtadmin administrator

授权远程访问（也可以登录后，可视化配置）

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

查看用户列表：

#查看用户列表
rabbitmqctl  list_users

修改对应用户密码

rabbitmqctl  change_password  username  'newpasswd'

6.5.2 界面方式：

创建用户并且超级管理员：

创建用户名hcxtuser，密码hcxtuser123!@#的用户

授权远程访问:

点击用户名称进行授权

6.5.3.Virtual Hosts配置

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

1-创建Virtual Hosts

如下图：

2-设置Virtual Hosts权限

6.6.常用配置说明：

tcp_listerners    #设置rabbimq的监听端口，默认为[5672]。
disk_free_limit     #磁盘低水位线，若磁盘容量低于指定值则停止接收数据，默认值为{mem_relative, 1.0},即与内存相关联1：1，也可定制为多少byte.
vm_memory_high_watermark    #设置内存低水位线，若低于该水位线，则开启流控机制，默认值是0.4，即内存总量的40%。
hipe_compile     #将部分rabbimq代码用High Performance Erlang compiler编译，可提升性能，该参数是实验性，若出现erlang vm segfaults，应关掉。
force_fine_statistics    #该参数属于rabbimq_management，若为true则进行精细化的统计，但会影响性能。
frame_max     #包大小，若包小则低延迟，若包则高吞吐，默认是131072=128K。
heartbeat     #客户端与服务端心跳间隔，设置为0则关闭心跳，默认是60秒。

7.RabbitMQ入门

7.1. 搭建示例工程

7.1.1. 创建工程

rabitmq-consumer （消费者）
rabitmq-producer （生产者）

7.1.2. 添加依赖

<dependencies>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.6.0</version>
    </dependency>
</dependencies>


<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.0</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

7.2. 编写生产者

package com.glj.rabitmq.simple;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 主机地址;默认为 localhost、127.0.0.1
        connectionFactory.setHost("127.0.0.1");
        // 连接端口;默认为 5672
        connectionFactory.setPort(5672);
        // 虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/hcxtuser");
        // 连接用户名；默认为guest
        connectionFactory.setUsername("hcxtuser");
        // 连接密码；默认为guest
        connectionFactory.setPassword("hcxtuser123!@#");

        try(
                // 创建连接
                Connection connection = connectionFactory.newConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            // 声明（创建）队列
            /**
             * 参数1：队列名称
             * 参数2：是否定义持久化队列
             * 参数3：是否独占本次连接
             * 参数4：是否在不使用的时候自动删除队列
             * 参数5：队列其它参数
             */
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);
            // 要发送的信息
            String mes = "hello rabitmq";
            /**
             * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
             * 参数2：路由key,简单模式可以传递队列名称
             * 参数3：消息其它属性
             * 参数4：消息内容
             */
            channel.basicPublish("", QUEUE_NAME, null, mes.getBytes());
            System.out.println("已发送消息：" + mes);


        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

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

7.3. 编写消费者

package com.glj.rabitmq.simple;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer {

    static final String QUEUE_NAME = "simple_queue";

    public static void main(String[] args) throws Exception {

        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 主机地址;默认为 localhost、127.0.0.1
        connectionFactory.setHost("127.0.0.1");
        // 连接端口;默认为 5672
        connectionFactory.setPort(5672);
        // 虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/hcxtuser");
        // 连接用户名；默认为guest
        connectionFactory.setUsername("hcxtuser");
        // 连接密码；默认为guest
        connectionFactory.setPassword("hcxtuser123!@#");

        // 创建连接
        Connection connection = connectionFactory.newConnection
                ();
        // 创建频道
        Channel channel = connection.createChannel();
        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //消息者标签
                System.out.println("consumerTag:"+consumerTag);
                //路由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, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

7.4. 小结

上述的入门案例中中其实使用的是如下的简单模式：

在上图的模型中，有以下概念：

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

8.AMQP

8.1. 相关概念介绍

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

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

RabbitMQ是AMQP协议的Erlang的实现。

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

8.1.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. 关闭连接；

9.RabbitMQ工作模式

9.1. Work queues工作队列模式

9.1.1. 模式说明

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

应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

9.1.2.代码

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

创建connection的工具类ConnectionUtil

package com.glj.rabitmq.util;

import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 21:09
 */
public class ConnectionUtil {

    public static Connection getConnection() throws Exception {
        // 创建连接工厂
        ConnectionFactory connectionFactory = new ConnectionFactory();
        // 主机地址;默认为 localhost、127.0.0.1
        connectionFactory.setHost("127.0.0.1");
        // 连接端口;默认为 5672
        connectionFactory.setPort(5672);
        // 虚拟主机名称;默认为 /
        connectionFactory.setVirtualHost("/hcxtuser");
        // 连接用户名；默认为guest
        connectionFactory.setUsername("hcxtuser");
        // 连接密码；默认为guest
        connectionFactory.setPassword("hcxtuser123!@#");

        //创建连接
        return connectionFactory.newConnection();
    }

}

生产者：

package com.glj.rabitmq.work;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            // 声明（创建）队列
            /**
             * 参数1：队列名称
             * 参数2：是否定义持久化队列
             * 参数3：是否独占本次连接
             * 参数4：是否在不使用的时候自动删除队列
             * 参数5：队列其它参数
             */
            channel.queueDeclare(QUEUE_NAME,true,false,false,null);

            for (int i = 1; i < 20; i++) {
                // 要发送的信息
                String mes = "rabitmq-work模式--"+i;
                /**
                 * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
                 * 参数2：路由key,简单模式可以传递队列名称
                 * 参数3：消息其它属性
                 * 参数4：消息内容
                 */
                channel.basicPublish("", QUEUE_NAME, null, mes.getBytes());
                System.out.println("已发送消息：" + mes);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

消费者1：

package com.glj.rabitmq.work;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

消费者2：

package com.glj.rabitmq.work;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    static final String QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws Exception {
        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();
        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(QUEUE_NAME,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(QUEUE_NAME,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

小结

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

9.2.订阅模式类型说明

前面2个案例中，只有3个角色：

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

订阅模式示例图：

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

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

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

9.3.Publish/Subscribe发布与订阅模式(fanout)

模式说明

发布订阅模式：
1、每个消费者监听自己的队列。
2、生产者将消息发给broker，由交换机将消息转发到绑定此交换机的每个队列，每个绑定交换机的队列都将接收
到消息

代码：

1.生产者：

package com.glj.rabitmq.ps;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "fanout_queue_2";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            /**
                声明交换机
                exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
               参数：
                1. exchange:交换机名称
                2. type:交换机类型
                    DIRECT("direct"),：定向
                    FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
                    TOPIC("topic"),通配符的方式
                    HEADERS("headers");参数匹配

                3. durable:是否持久化
                4. autoDelete:自动删除
                5. internal：内部使用。 一般false
                6. arguments：参数
             */
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,false,null);

            // 声明（创建）队列
            /**
             * 参数1：队列名称
             * 参数2：是否定义持久化队列
             * 参数3：是否独占本次连接
             * 参数4：是否在不使用的时候自动删除队列
             * 参数5：队列其它参数
             */
            channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
            channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);

            /**
             * 队列绑定交换机
            queueBind(String queue, String exchange, String routingKey)
            参数：
                1. queue：队列名称
                2. exchange：交换机名称
                3. routingKey：路由键，绑定规则
                    如果交换机的类型为fanout ，routingKey设置为""
             */
            channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"");

            for (int i = 1; i < 20; i++) {
                // 要发送的信息
                String mes = "rabitmq-work模式--"+i;
                /**
                 * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
                 * 参数2：路由key,简单模式可以传递队列名称
                 * 参数3：消息其它属性
                 * 参数4：消息内容
                 */
                channel.basicPublish(FANOUT_EXCHAGE, "", null, mes.getBytes());
                System.out.println("已发送消息：" + mes);
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

2.消费者1:

package com.glj.rabitmq.ps;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_1";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数：
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),：定向
         FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal：内部使用。 一般false
         6. arguments：参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,false,null);

        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数：
         1. queue：队列名称
         2. exchange：交换机名称
         3. routingKey：路由键，绑定规则
         如果交换机的类型为fanout ，routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

3.消费者2：

package com.glj.rabitmq.ps;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "fanout_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "fanout_queue_2";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数：
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),：定向
         FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal：内部使用。 一般false
         6. arguments：参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.FANOUT,true,false,false,null);

        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数：
         1. queue：队列名称
         2. exchange：交换机名称
         3. routingKey：路由键，绑定规则
         如果交换机的类型为fanout ，routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

4.小结

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

5.发布订阅模式与工作队列模式的区别

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

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

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

9.4. Routing路由模式(direct)

模式说明

路由模式特点：

队列与交换机的绑定，不能是任意绑定了，而是要指定一个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 的消息

代码

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

1.生产者：

package com.glj.rabitmq.routing;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    //交换机名称
    static final String FANOUT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "direct_queue_1";
    //队列名称
    static final String FANOUT_QUEUE_2 = "direct_queue_2";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            /**
                声明交换机
                exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
               参数：
                1. exchange:交换机名称
                2. type:交换机类型
                    DIRECT("direct"),：定向
                    FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
                    TOPIC("topic"),通配符的方式
                    HEADERS("headers");参数匹配

                3. durable:是否持久化
                4. autoDelete:自动删除
                5. internal：内部使用。 一般false
                6. arguments：参数
             */
            channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,false,null);

            // 声明（创建）队列
            /**
             * 参数1：队列名称
             * 参数2：是否定义持久化队列
             * 参数3：是否独占本次连接
             * 参数4：是否在不使用的时候自动删除队列
             * 参数5：队列其它参数
             */
            channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);
            channel.queueDeclare(FANOUT_QUEUE_2,true,false,false,null);

            /**
             * 队列绑定交换机
            queueBind(String queue, String exchange, String routingKey)
            参数：
                1. queue：队列名称
                2. exchange：交换机名称
                3. routingKey：路由键，绑定规则
                    如果交换机的类型为fanout ，routingKey设置为""
             */
            channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"error");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"error");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"info");
            channel.queueBind(FANOUT_QUEUE_2,FANOUT_EXCHAGE,"warn");

            // 要发送的信息
            String mes = "rabitmq---info";
            /**
             * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
             * 参数2：路由key,简单模式可以传递队列名称
             * 参数3：消息其它属性
             * 参数4：消息内容
             */
            channel.basicPublish(FANOUT_EXCHAGE, "info", null, mes.getBytes());
            System.out.println("已发送消息：" + mes);

            // 要发送的信息
            mes = "rabitmq---error";
            channel.basicPublish(FANOUT_EXCHAGE, "error", null, mes.getBytes());
            System.out.println("已发送消息：" + mes);
        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

2.消费者1：

package com.glj.rabitmq.routing;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "direct_queue_1";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数：
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),：定向
         FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal：内部使用。 一般false
         6. arguments：参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,false,null);

        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数：
         1. queue：队列名称
         2. exchange：交换机名称
         3. routingKey：路由键，绑定规则
         如果交换机的类型为fanout ，routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"error");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

3.消费者2：

package com.glj.rabitmq.routing;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    //交换机名称
    static final String FANOUT_EXCHAGE = "direct_exchange";
    //队列名称
    static final String FANOUT_QUEUE_1 = "direct_queue_2";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数：
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),：定向
         FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal：内部使用。 一般false
         6. arguments：参数
         */
        channel.exchangeDeclare(FANOUT_EXCHAGE, BuiltinExchangeType.DIRECT,true,false,false,null);

        // 声明（创建）队列
        /**
         * 参数1：队列名称
         * 参数2：是否定义持久化队列
         * 参数3：是否独占本次连接
         * 参数4：是否在不使用的时候自动删除队列
         * 参数5：队列其它参数
         */
        channel.queueDeclare(FANOUT_QUEUE_1,true,false,false,null);

        /**
         * 队列绑定交换机
         queueBind(String queue, String exchange, String routingKey)
         参数：
         1. queue：队列名称
         2. exchange：交换机名称
         3. routingKey：路由键，绑定规则
         如果交换机的类型为fanout ，routingKey设置为""
         */
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"error");
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"info");
        channel.queueBind(FANOUT_QUEUE_1,FANOUT_EXCHAGE,"warn");

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者2-接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(FANOUT_QUEUE_1,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

小结

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

9.5.Topics通配符模式

模式说明

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

Routingkey 一般都是有一个或多个单词组成，多个单词之间以”.”分割，例如： item.insert

通配符规则：

#：匹配一个或多个词

*：匹配不多不少恰好1个词

举例：

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

item.*：只能匹配item.insert

图解：

红色Queue：绑定的是usa.# ，因此凡是以 usa.开头的routing key 都会被匹配到
黄色Queue：绑定的是#.news ，因此凡是以 .news结尾的 routing key 都会被匹配

代码

在编码上与 路由模式 的区别是交换机的类型为：TOPIC，还有消费者不需要绑定声明队列、不需要队列绑定交换机。

1.生产者

使用topic类型的Exchange，发送消息的routing key有3种： #.error、user.*、*.*：

package com.glj.rabitmq.topic;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:16
 */
public class Producer {

    //交换机名称
    static final String EXCHAGE = "topic_exchange";
    //队列名称
    static final String QUEUE_1 = "topic_queue_1";
    //队列名称
    static final String QUEUE_2 = "topic_queue_2";

    public static void main(String[] args) {
        try(
                // 创建连接
                Connection connection = ConnectionUtil.getConnection();
                // 创建频道
                Channel channel = connection.createChannel()
        ){
            /**
                声明交换机
                exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
               参数：
                1. exchange:交换机名称
                2. type:交换机类型
                    DIRECT("direct"),：定向
                    FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
                    TOPIC("topic"),通配符的方式
                    HEADERS("headers");参数匹配

                3. durable:是否持久化
                4. autoDelete:自动删除
                5. internal：内部使用。 一般false
                6. arguments：参数
             */
            channel.exchangeDeclare(EXCHAGE, BuiltinExchangeType.TOPIC,true,false,false,null);

            // 声明（创建）队列
            /**
             * 参数1：队列名称
             * 参数2：是否定义持久化队列
             * 参数3：是否独占本次连接
             * 参数4：是否在不使用的时候自动删除队列
             * 参数5：队列其它参数
             */
            channel.queueDeclare(QUEUE_1,true,false,false,null);
            channel.queueDeclare(QUEUE_2,true,false,false,null);

            /**
             * 队列绑定交换机
            queueBind(String queue, String exchange, String routingKey)
            参数：
                1. queue：队列名称
                2. exchange：交换机名称
                3. routingKey：路由键，绑定规则
                    如果交换机的类型为fanout ，routingKey设置为""
             */
            channel.queueBind(QUEUE_1,EXCHAGE,"#.error");
            channel.queueBind(QUEUE_2,EXCHAGE,"user.*");
            channel.queueBind(QUEUE_2,EXCHAGE,"*.*");

            // 要发送的信息
            String mes = "rabitmq---info";
            /**
             * 参数1：交换机名称，如果没有指定则使用默认Default Exchage
             * 参数2：路由key,简单模式可以传递队列名称
             * 参数3：消息其它属性
             * 参数4：消息内容
             */
            channel.basicPublish(EXCHAGE, "user.info", null, mes.getBytes());
            System.out.println("已发送消息：" + mes);

        }catch (Exception e){
            e.printStackTrace();
        }
    }

}

2.消费者:one::

package com.glj.rabitmq.topic;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer1 {

    //交换机名称
    static final String EXCHAGE = "topic_exchange";
    //队列名称
    static final String QUEUE_1 = "topic_queue_1";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数：
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),：定向
         FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal：内部使用。 一般false
         6. arguments：参数
         */
        channel.exchangeDeclare(EXCHAGE, BuiltinExchangeType.TOPIC,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(QUEUE_1,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

3.消费者:two::

package com.glj.rabitmq.topic;

import com.glj.rabitmq.util.ConnectionUtil;
import com.rabbitmq.client.*;

import java.io.IOException;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/20
 * @Time: 20:44
 */
public class Consumer2 {

    //交换机名称
    static final String EXCHAGE = "topic_exchange";
    //队列名称
    static final String QUEUE_1 = "topic_queue_2";

    public static void main(String[] args) throws Exception {

        // 创建连接
        Connection connection = ConnectionUtil.getConnection();
        // 创建频道
        Channel channel = connection.createChannel();

        /**
         声明交换机
         exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
         参数：
         1. exchange:交换机名称
         2. type:交换机类型
         DIRECT("direct"),：定向
         FANOUT("fanout"),：扇形（广播），发送消息到每一个与之绑定队列。
         TOPIC("topic"),通配符的方式
         HEADERS("headers");参数匹配

         3. durable:是否持久化
         4. autoDelete:自动删除
         5. internal：内部使用。 一般false
         6. arguments：参数
         */
        channel.exchangeDeclare(EXCHAGE, BuiltinExchangeType.TOPIC,true,false,false,null);

        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            @Override
            /**
             回调方法，当介绍消息后，会自动执行该方法
             * consumerTag 消息者标签，在channel.basicConsume时候可以指定
             * envelope 消息包的内容，可从中获取消息id，消息routingkey，交换机，消息和重传标志(收到消息失败后是否需要重新发送)
             * properties 属性信息
             * body 消息
             */
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                //收到的消息
                System.out.println("消费者1-接收到的消息为：" + new String(body, "utf-8"));

            }
        };
        //监听消息
        /**
         * 参数1：队列名称
         * 参数2：是否自动确认，设置为true为表示消息接收到自动向mq回复接收到了，mq接收到回复会删除消息，设置为false则需要手动确认
         * 参数3：消息接收到后回调
         */
        channel.basicConsume(QUEUE_1,true,defaultConsumer);

        //不关闭资源，应该一直监听消息
        //channel.close();
        //connection.close();
    }

}

小结

Topic主题模式可以实现 Publish/Subscribe发布与订阅模式 和 Routing路由模式 的功能；只是Topic在配置routing key 的时候可以使用通配符，显得更加灵活。

9.6. 模式总结

RabbitMQ工作模式：

1、简单模式 HelloWorld

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）

2、工作队列模式 Work Queue

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

3、发布订阅模式 Publish/subscribe

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

4、路由模式 Routing

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

5、通配符模式 Topic

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

10.Spring Boot整合RabbitMQ

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

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

一般在开发过程中：

生产者工程：

application.properties文件配置RabbitMQ相关信息；
在生产者工程中编写配置类，用于创建交换机和队列，并进行绑定
注入RabbitTemplate对象，通过RabbitTemplate对象发送消息到交换机

消费者工程：

application.properties文件配置RabbitMQ相关信息
创建消息处理类，用于接收队列中的消息并进行处理

10.1 搭建生产者工程

创建生产者工程rabitmq-producer-boot

pom.xml 引入：

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

10.1.1配置RabbitMQ

配置文件application.properties

spring.rabbitmq.host=127.0.0.1
spring.rabbitmq.port=5672
spring.rabbitmq.username=hcxtuser
spring.rabbitmq.password=hcxtuser123!@#
spring.rabbitmq.virtual-host=/hcxtuser

绑定交换机和队列

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

package com.glj.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/22
 * @Time: 0:02
 */
@Configuration
public class RabbitMQConfig {

    //交换机名称
    public static final String EXCHANGE_NAME = "boot_topic_exchange";
    //队列名称
    public static final String QUEUE_NAME = "boot_queue";

    //声明交换机
    @Bean("bootExchange")
    public Exchange bootExchange(){
        return ExchangeBuilder.topicExchange(EXCHANGE_NAME).durable(true).build();
    }

    //声明队列
    @Bean("bootQueue")
    public Queue bootQueue(){
        return QueueBuilder.durable(QUEUE_NAME).build();
    }

    //绑定队列和交换机
    @Bean
    public Binding bootQueueExchange(@Qualifier("bootQueue") Queue queue,
                                     @Qualifier("bootExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("boot.#").noargs();
    }

}

测试.

package com.glj;

import com.glj.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 RabitmqProducerBootApplicationTests {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Test
    public void testSend() {
        String msg = "此消息:boot.rabbitmq";
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.rabbitmq",msg);
    }

}

10.2 搭建消费者工程

pom.xml 和application.properties与生产者一致…

10.2.1配置RabbitMQ

消息监听处理类

编写消息监听器com.glj.rabbitmq.listener.RabbitMQListener

package com.glj.rabbitmq.listener;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/22
 * @Time: 0:23
 */

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

@Component
public class RabbitMQListener {

    @RabbitListener(queues = "boot_queue")
    public void ListenQueue(Message message){
        System.out.println("消费者接收到的消息为：" + new String(message.getBody()));
    }

}

11.RabbitMQ 高级特性

11.1 消息的可靠投递

在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。

confirm 确认模式
return 退回模式

rabbitmq 整个消息投递的路径为：

producer—>rabbitmq broker—>exchange—>queue—>consumer

消息从 producer 到 exchange 则会返回一个 confirmCallback 。
消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。

我们将利用这两个 callback 控制消息的可靠性投递

confirm 确认模式

消息从 producer 到 exchange 则会返回一个 confirmCallback

在spring boot 生产者中继续咱们的操作.

1.确认模式的开启

在配置文件application.properties 新增 spring.rabbitmq.publisher-confirms=true

2.在rabbitTemplate 定义ConfirmCallBack回调函数

package com.glj.rabbitmq.config;

import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * Created with IntelliJ IDEA.
 *
 * @Author: GuoLiangJun
 * @Date: 2021/6/22
 * @Time: 0:02
 */
@Configuration
public class RabbitMQConfigWithConfirm {

    public static final String CONFIRM_EXCHANGE_NAME = "confirm_exchange";
    public static final String CONFIRM_QUEUE_NAME = "confirm_queue";


    //确认模式

    @Bean("confirmExchange")
    public Exchange confirmExchange(){
        return new TopicExchange(CONFIRM_EXCHANGE_NAME,true,false,null);
    }

    @Bean("confirmQueue")
    public Queue confirmQueue(){
        return new Queue(CONFIRM_QUEUE_NAME,true,false,false);
    }

    @Bean
    public Binding confirmQueueExchange(@Qualifier("confirmQueue") Queue queue,
                                        @Qualifier("confirmExchange") Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with("confirm").noargs();
    }

}

测试类中：

@Test
public void testConfirm(){
    // 定义回调
    rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
        /**
             *
             * @param correlationData 相关配置信息
             * @param ack exchange交换机 是否成功收到消息
             * @param cause 失败原因
             */
        @Override
        public void confirm(CorrelationData correlationData, boolean ack, String cause) {
            System.out.println("confirm方法被执行了..."+ack);
            if (!ack){
                //接受成功
                System.out.println("接受"+cause);
            }else{
                //失败
                System.out.println("失败原因："+cause);
            }
        }
    });
    String msg = "此消息:confirm 方法测试";
    rabbitTemplate.convertAndSend(RabbitMQConfigWithConfirm.CONFIRM_EXCHANGE_NAME,"confirm",msg);
}

return 退回模式

消息从 exchange–>queue 投递失败则会返回一个 returnCallback 。

在spring boot 生产者中继续咱们的操作.

1.确认模式的开启

在配置文件application.properties 新增 spring.rabbitmq.publisher-returns=true

2.在rabbitTemplate 定义ReturnCallBack回调函数

@Test
public void testReturn(){
    //设置交换机处理失败消息的模式
    rabbitTemplate.setMandatory(true);

    //设置回调
    rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
        /**
             *
             * @param message 消息对象
             * @param replyCode 返回错误码
             * @param replyText 返回错误信息
             * @param exchange 交换机
             * @param routingKey 路由
             */
        @Override
        public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println("return 执行了....");
            System.out.println("message："+message);
            System.out.println("replyCode："+replyCode);
            System.out.println("replyText："+replyText);
            System.out.println("exchange："+exchange);
            System.out.println("routingKey："+routingKey);
        }
    });
    String msg = "此消息:confirm-return 方法测试";
    rabbitTemplate.convertAndSend("confirm_exchange","confirm111",msg);
}

消息的可靠投递小结

设置ConnectionFactory的publisher-confirms=”true” 开启确认模式。

使用rabbitTemplate.setConfirmCallback设置回调函数。当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理。

设置ConnectionFactory的publisher-returns=”true” 开启退回模式。

使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到queue失败后，如果设置了rabbitTemplate.setMandatory(true)参数，则会将消息退回给producer。并执行回调函数returnedMessage。

在RabbitMQ中也提供了事务机制，但是性能较差，此处不做讲解。使用channel下列方法，完成事务控制：

txSelect(), 用于将当前channel设置成transaction模式
txCommit()，用于提交事务
txRollback(),用于回滚事务

Consumer Ack

ack指Acknowledge，确认。表示消费端收到消息后的确认方式。

有三种确认方式：

自动确认：acknowledge=”none”
手动确认：acknowledge=”manual”
根据异常情况确认：acknowledge=”auto”

其中自动确认是指，当消息一旦被Consumer接收到，则自动确认收到，并将相应 message 从 RabbitMQ 的消息缓存中移除。但是在实际业务处理中，很可能消息接收到，业务处理出现异常，那么该消息就会丢失。如果设置了手动确认方式，则需要在业务处理成功后，调用channel.basicAck()，手动签收，如果出现异常，则调用channel.basicNack()方法，让其自动重新发送消息。

消费端限流

Ø配置 prefetch属性设置消费端一次拉取多少消息

Ø消费端的确认模式一定为手动确认。acknowledge=”manual”

TTL

TTL 全称 Time To Live（存活时间/过期时间）。

当消息到达存活时间后，还没有被消费，会被自动清除。

RabbitMQ可以对消息设置过期时间，也可以对整个队列（Queue）设置过期时间。

实现：

设置队列过期时间使用参数：x-message-ttl，单位：ms(毫秒)，会对整个队列消息统一过期。

设置消息过期时间使用参数：expiration。单位：ms(毫秒)，当该消息在队列头部时（消费时），会单独判断这一消息是否过期。

如果两者都进行了设置，以时间短的为准。

死信队列

死信队列，英文缩写：DLX 。Dead Letter Exchange（死信交换机），当消息成为Dead message后，可以被重新发送到另一个交换机，这个交换机就是DLX。

消息成为死信的三种情况：

队列消息长度到达限制；
消费者拒接消费消息，basicNack/basicReject,并且不把消息重新放入原目标队列,requeue=false；
原队列存在消息过期设置，消息到达超时时间未被消费；

队列绑定死信交换机：

给队列设置参数： ead-letter-exchange 和 ead-letter-routing-key

死信队列小结

1.死信交换机和死信队列和普通的没有区别

2.当消息成为死信后，如果该队列绑定了死信交换机，则消息会被死信交换机重新路由到死信队列

3.消息成为死信的三种情况：

队列消息长度到达限制；
消费者拒接消费消息，并且不重回队列；
原队列存在消息过期设置，消息到达超时时间未被消费；

延迟队列

延迟队列，即消息进入队列后不会立即被消费，只有到达指定时间后，才会被消费。

使用：TTL+死信队列组合实现延迟队列的效果。

延迟队列小结

延迟队列指消息进入队列后，可以被延迟一定时间，再进行消费。
RabbitMQ没有提供延迟队列功能，但是可以使用： TTL + DLX 来实现延迟队列效果。