消息中间件RabbitMQ（二）之Java客户端操作

在上篇【消息中间件RabbitMQ（一）之初见】中已经了解了 RabbitMQ 的安装、相关概念以及图形化管理界面的使用，接下来学习一下使用 Java 客户端来操作 RabbitMQ。

入门程序

依赖

1、使用 Maven 创建生产者工程和消费者 Java 工程，分别引入 RabbitMQ 客户端依赖：

<dependency>
    <groupId>com.rabbitmq</groupId>
    <artifactId>amqp-client</artifactId>
    <version>4.0.3</version> <!-- 此版本与 Spring Boot 1.5.9 版本匹配 -->
</dependency>

生产者

2、编写生产者程序：

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

/**
 * RabbitMQ 入门程序-生产者
 */
public class Producer01 {
    /**
     * 队列名称
     */
    private static final String QUEUE_NAME = "helloworld";

    public static void main(String[] args) throws Exception {
        //  通过连接工厂创建和 mq 服务的连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.146.131"); // mq 服务主机地址
        connectionFactory.setPort(5672);        // mq 服务暴露的端口
        connectionFactory.setUsername("guest"); // 用户名
        connectionFactory.setPassword("guest"); // 密码
        connectionFactory.setVirtualHost("/");  // 设置虚拟机路径
        // 一个 mq 的服务可以设置多个虚拟机路径，每个虚拟机路径对应一个虚拟机（相当于一个独立的 mq 服务），即每个虚拟机路径下的 mq 服务环境是相互隔离的。
        Connection connection = null;
        Channel channel = null;
        try {
            // 创建新连接
            connection = connectionFactory.newConnection();
            // 创建会话通道（channel），客户端和 mq 服务都是通过连接中的会话通道通信
            channel = connection.createChannel();
            // 声明队列，如果队列不存在，则创建
            /**
             * 参数列表：
             *  String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *      queue：队列名称
             *      durable：是否持久化，mq 重启后队列依旧存在
             *      exclusive：是否排他，独占该连接，队列仅能被此连接访问，如果连接关闭，则队列自动删除
             *      autoDelete：自动删除，队列不再使用时自动删除此队列，如果和 exclusive 都为 true，则定义的队列就是临时队列
             *      arguments：参数，可设置队列的扩展参数，如设置队列的存活时间等...
             */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            // 发布消息
            /**
             * 参数列表：
             *  String exchange, String routingKey, BasicProperties props, byte[] body
             *      exchange：交换机，如果不指定则使用默认的交换机
             *      routingKey：路由键，交换机按路由键来将消息转发到指定的队列，如果使用默认交换机，那么 routingKey 要设为队列的名称
             *      props：消息属性
             *      body：消息体，消息内容
             */
            String msg = "hello world";
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.printf("Send msg " + msg + " to " + QUEUE_NAME);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) channel.close();
            if (connection != null) connection.close();
        }
    }
}

执行该生产者程序，执行完毕后可通过管理界面查看到队列及消息信息：

生产者发送消息流程小结：

1. 创建连接；
2. 创建通道；
3. 声明队列；
4. 发送消息；

消费者

3、编写消费者程序：

import com.rabbitmq.client.*;

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

/**
 * RabbitMQ 入门程序-消费者
 */
public class Consumer01 {

    private static final String QUEUE_NAME = "helloworld";

    public static void main(String[] args) throws Exception {
        //  通过连接工厂创建和 mq 服务的连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.146.131"); // mq 服务主机地址
        connectionFactory.setPort(5672);        // mq 服务暴露的端口
        connectionFactory.setUsername("guest"); // 用户名
        connectionFactory.setPassword("guest"); // 密码
        connectionFactory.setVirtualHost("/");  // 设置虚拟机路径
        // 一个 mq 的服务可以设置多个虚拟机路径，每个虚拟机路径对应一个虚拟机（相当于一个独立的 mq 服务），即每个虚拟机路径下的 mq 服务环境是相互隔离的。
        Connection connection = null;
        Channel channel = null;
        try {
            // 创建新连接
            connection = connectionFactory.newConnection();
            // 创建会话通道（channel），客户端和 mq 服务都是通过连接中的会话通道通信
            channel = connection.createChannel();
            // 声明队列，如果队列不存在，则创建
            /**
             * 参数列表：
             *  String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *      queue：队列名称
             *      durable：是否持久化，mq 重启后队列依旧存在
             *      exclusive：是否排他，独占该连接，队列仅能被此连接访问，如果连接关闭，则队列自动删除
             *      autoDelete：自动删除，队列不再使用时自动删除此队列，如果和 exclusive 都为 true，则定义的队列就是临时队列
             *      arguments：参数，可设置队列的扩展参数，如设置队列的存活时间等...
             */
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
            // 监听队列接收（消费）消息
            /**
             * 参数列表：
             *  String queue, boolean autoAck, String consumerTag, Consumer callback
             *      queue：监听的队列名称
             *      autoAck：自动回复，当消费者接收到消息后要告诉 mq 消息已接收，如果将此参数设置为 true 那么将会自动回复 mq，设置为 false 时需要手动编程回复 mq
             *      consumerTag：消费者标签，消费者的标识，可选
             *      callback：接收到消息时的回调对象
             */
            channel.basicConsume(QUEUE_NAME, true,"Consumer01", new DefaultConsumer(channel){
                /**
                 * 接收到消息时该方法将被调用
                 * @param consumerTag 消费者标签，消费者标识，使用 basicConsume 方法时使用的标签名称
                 * @param envelope 信封，存放了消息的相关信息，如发送该消息的交换机、该消息在 mq 中的标识
                 * @param properties 消息属性，生产者使用 channel.basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body) 发送消息时传递的 props 属性信息
                 * @param body 消息内容
                 * @throws IOException
                 */
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    String exchange = envelope.getExchange(); // 获取发送消息的交换机
                    long deliveryTag = envelope.getDeliveryTag(); // 消息在 mq 中的标识 id，可用于消息接收确认
                    System.out.println("receive msg from "+ consumerTag + "："+ new String(body, "utf-8"));
                }
            });

        } catch (Exception e) {
            e.printStackTrace();
        }
        // 消费者不关闭连接则持续监听接收（消费）队列中的消息
        //finally {
        //    if (channel != null) channel.close();
        //    if (connection != null) connection.close();
        //}
    }
}

执行该生产者程序，执行完毕后可通过管理界面查看到队列中的消息已被消费（接收），且之后生产者每发送一条消息到队列，消费者也随之消费一条消息；
消费者接收消息流程小结：

1. 创建连接；
2. 创建通道；
3. 声明队列（如果能够确定队列存在，则可忽略）；
4. 监听队列接收消息；
5. 回复 mq（如果设置了自动回复可忽略）；

工作模式

RabbitMQ 有如下几种工作模式：

1. Simple：简单模式；
2. Work queues：工作队列模式；
3. Publish/Subscribe：发布订阅模式；
4. Routing：路由模式；
5. Topics：通配符模式；
6. Header：Header 转发器模式；
7. RPC：远程过程调用模式；

各个工作模式使用的交换机类型可能是不一样的，如下：

• Fanout Exchange 对应发布订阅模式；
• Direct Exchange 对应路由模式；
• Topic Exchange 对应通配符模式；
• Headers Exchange 对应 Header 转发器模式；

上述交换机类型都定义在 com.rabbitmq.client.BuiltinExchangeType 枚举类中。

Simple 简单模式

生产者将消息放入队列， 消费者（Consumer) 监听消息队列，如果队列中有消息，就消费掉，消息被拿走后，消息自动从队列中删除。

隐患：消息可能没有被消费者正确处理，但该消息已经从队列中消失了，造成消息的丢失。
应用场景：聊天。

示例

上面入门程序使用的就是简单模式。

Work queues 工作队列模式

生产者将消息放入队列，消费者可以有多个。如消费者 C1 、消费者 C2 同时监听同一个队列。
C1 和 C2 共同消费当前的消息队列内容，Rabbit MQ 默认使用轮询的方式将消息平均发送给各个消费者。
上图中虽然没有出现交换机，但实际上是使用了默认的交换机。

隐患：高并发情况下，可能会产生某一个消息被多个消费者共同使用，此时可以设置同步，保证一条消息只能被一个消费者使用。
应用场景：抢红包，大项目中的资源调度（任务分配系统不需知道哪一个任务执行系统在空闲，直接将任务扔到消息队列中，空闲的系统自动争抢）。

示例

将入门程序中的消费者程序启动多个，则就是工作队列模式。

publish/subscribe 发布订阅模式

X 代表交换机，是 RabbitMQ 内部组件,生产者将消息放入交换机，交换机负责把消息发送到所有绑定了该交换机的队列，对应消息队列的消费者拿到消息进行消费。

应用场景：邮件群发、群聊天、广播、广告。

示例

编写生产者程序：

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

/**
 * RabbitMQ 发布订阅模式-生产者
 */
public class Producer02 {
    private static final String FANOUT_EXCHANGE = "test_fanout_exchange";
    private static final String QUEUE_1 = "queue_1";
    private static final String QUEUE_2 = "queue_2";

    public static void main(String[] args) throws Exception {
        //  通过连接工厂创建和 mq 服务的连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.146.131"); // mq 服务主机地址
        connectionFactory.setPort(5672);        // mq 服务暴露的端口
        connectionFactory.setUsername("guest"); // 用户名
        connectionFactory.setPassword("guest"); // 密码
        connectionFactory.setVirtualHost("/");  // 设置虚拟机路径
        // 一个 mq 的服务可以设置多个虚拟机路径，每个虚拟机路径对应一个虚拟机（相当于一个独立的 mq 服务），即每个虚拟机路径下的 mq 服务环境是相互隔离的。
        Connection connection = null;
        Channel channel = null;
        try {
            // 创建新连接
            connection = connectionFactory.newConnection();
            // 创建会话通道（channel），客户端和 mq 服务都是通过连接中的会话通道通信
            channel = connection.createChannel();
            // 声明队列，如果队列不存在，则创建
            /**
             * 参数列表：
             *  String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *      queue：队列名称
             *      durable：是否持久化，mq 重启后队列依旧存在
             *      exclusive：是否排他，独占该连接，队列仅能被此连接访问，如果连接关闭，则队列自动删除
             *      autoDelete：自动删除，队列不再使用时自动删除此队列，如果和 exclusive 都为 true，则定义的队列就是临时队列
             *      arguments：参数，可设置队列的扩展参数，如设置队列的存活时间等...
             */
            channel.queueDeclare(QUEUE_1, true, false, false, null);
            channel.queueDeclare(QUEUE_2, true, false, false, null);
            // 声明交换机
            /**
             * 参数列表：
             *  String exchange, String type
             *      exchange：交换机名称
             *      type：交换机类型，具体信息可参考上篇文章中的 Exchange 类型
             */
            channel.exchangeDeclare(FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT);

            // 绑定交换机和队列
            /**
             * 参数列表：
             *  String queue, String exchange, String routingKey
             *      queue：队列名称
             *      exchange：交换机名称
             *      routingKey：路由 key，在发布订阅模式中无用，设为 "" 即可
             */
            channel.queueBind(QUEUE_1, FANOUT_EXCHANGE, "");
            channel.queueBind(QUEUE_2, FANOUT_EXCHANGE, "");

            // 发布消息给交换机
            /**
             * 参数列表：
             *  String exchange, String routingKey, BasicProperties props, byte[] body
             *      exchange：交换机，如果不指定则使用默认的交换机
             *      routingKey：路由键，交换机按路由键来将消息转发到指定的队列，如果使用默认交换机，那么 routingKey 要设为队列的名称
             *      props：消息属性
             *      body：消息体，消息内容
             */
            String msg = "hello world";
            channel.basicPublish(FANOUT_EXCHANGE, "", null, msg.getBytes());
            System.out.printf("Send msg " + msg + " to " + FANOUT_EXCHANGE);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) channel.close();
            if (connection != null) connection.close();
        }
    }
}

发布订阅模式对应的交换机类型为 Fanout，所以该生产者程序在入门程序的基础上添加了手动定义的 Fanout 类型的交换机，并且声明了 2 个队列绑定到了该交换机。
执行该程序，查看会发现两个队列都接受到了消息：

Routing 路由模式

生产者将消息携带一个路由 key（字符串） 发送给交换机，交换机根据路由 key，匹配到与该路由 key 对应的消息队列，对应消息队列的消费者消费消息。

应用场景：消息分类通知（利用路由 key 可将消息分类）。

示例

编写生产者程序：

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

/**
 * RabbitMQ 路由模式-生产者
 */
public class Producer03 {
    private static final String DIRECT_EXCHANGE = "test_direct_exchange";
    private static final String QUEUE_3 = "queue_3";
    private static final String QUEUE_4 = "queue_4";

    /**
     * queue_3 使用的路由 key
     */
    private static final String QUEUE_3_ROUTING_KEY = "queue_3_key";
    /**
     * queue_4 使用的路由 key
     */
    private static final String QUEUE_4_ROUTING_KEY = "queue_4_key";

    public static void main(String[] args) throws Exception {
        //  通过连接工厂创建和 mq 服务的连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.146.131"); // mq 服务主机地址
        connectionFactory.setPort(5672);        // mq 服务暴露的端口
        connectionFactory.setUsername("guest"); // 用户名
        connectionFactory.setPassword("guest"); // 密码
        connectionFactory.setVirtualHost("/");  // 设置虚拟机路径
        // 一个 mq 的服务可以设置多个虚拟机路径，每个虚拟机路径对应一个虚拟机（相当于一个独立的 mq 服务），即每个虚拟机路径下的 mq 服务环境是相互隔离的。
        Connection connection = null;
        Channel channel = null;
        try {
            // 创建新连接
            connection = connectionFactory.newConnection();
            // 创建会话通道（channel），客户端和 mq 服务都是通过连接中的会话通道通信
            channel = connection.createChannel();
            // 声明队列，如果队列不存在，则创建
            /**
             * 参数列表：
             *  String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *      queue：队列名称
             *      durable：是否持久化，mq 重启后队列依旧存在
             *      exclusive：是否排他，独占该连接，队列仅能被此连接访问，如果连接关闭，则队列自动删除
             *      autoDelete：自动删除，队列不再使用时自动删除此队列，如果和 exclusive 都为 true，则定义的队列就是临时队列
             *      arguments：参数，可设置队列的扩展参数，如设置队列的存活时间等...
             */
            channel.queueDeclare(QUEUE_3, true, false, false, null);
            channel.queueDeclare(QUEUE_4, true, false, false, null);
            // 声明交换机
            /**
             * 参数列表：
             *  String exchange, String type
             *      exchange：交换机名称
             *      type：交换机类型，具体信息可参考上篇文章中的 Exchange 类型
             */
            channel.exchangeDeclare(DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT);

            // 绑定交换机和队列
            /**
             * 参数列表：
             *  String queue, String exchange, String routingKey
             *      queue：队列名称
             *      exchange：交换机名称
             *      routingKey：路由 key
             */
            channel.queueBind(QUEUE_3, DIRECT_EXCHANGE, QUEUE_3_ROUTING_KEY);
            channel.queueBind(QUEUE_4, DIRECT_EXCHANGE, QUEUE_4_ROUTING_KEY);

            // 发布消息给交换机
            /**
             * 参数列表：
             *  String exchange, String routingKey, BasicProperties props, byte[] body
             *      exchange：交换机，如果不指定则使用默认的交换机
             *      routingKey：路由键，交换机按路由键来将消息转发到指定的队列，如果使用默认交换机，那么 routingKey 要设为队列的名称
             *      props：消息属性
             *      body：消息体，消息内容
             */
            String routingKey = "";
            for (int i = 0; i < 10; i++) {
                String msg = "hello world" + i;
                routingKey = i % 2 == 0 ? QUEUE_3_ROUTING_KEY : QUEUE_4_ROUTING_KEY;
                channel.basicPublish(DIRECT_EXCHANGE, routingKey, null, msg.getBytes());
                System.out.println("Send msg " + msg + " to " + DIRECT_EXCHANGE);
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) channel.close();
            if (connection != null) connection.close();
        }
    }
}

路由模式对应的交换机类型为 Direct，所以该生产者程序在入门程序的基础上添加了手动定义的 Direct 类型的交换机，并且声明了 2 个队列绑定到了该交换机，在发布消息时也是直接发送给交换机，但发送消息的同时指定了路由 key 让消息分发到两个队列。
执行该程序，查看会发现两个队列都接受到了消息：

Topics 通配符模式

符号 * 和 # 都是通配符，# 可匹配一个或多个词，* 匹配单个词，每个单词间使用 . 分隔。
实际上就是在路由模式的基础上添加了模糊匹配功能，生产者把消息携带一个带通配符的路由 key 交给交换机，交换机根据路由 key 的规则模糊匹配到对应的队列，由对应队列的消费者消费消息；

示例

编写生产者程序：

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

/**
 * RabbitMQ 通配符模式-生产者
 */
public class Producer04 {
    private static final String TOPIC_EXCHANGE = "test_topic_exchange";
    private static final String QUEUE_5 = "queue_5";
    private static final String QUEUE_6 = "queue_6";

    /**
     * queue_5 使用的路由 key
     */
    private static final String QUEUE_5_ROUTING_KEY = "msg.#.queue_5.#";
    /**
     * queue_6 使用的路由 key
     */
    private static final String QUEUE_6_ROUTING_KEY = "msg.#.queue_6.#";

    public static void main(String[] args) throws Exception {
        //  通过连接工厂创建和 mq 服务的连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.146.131"); // mq 服务主机地址
        connectionFactory.setPort(5672);        // mq 服务暴露的端口
        connectionFactory.setUsername("guest"); // 用户名
        connectionFactory.setPassword("guest"); // 密码
        connectionFactory.setVirtualHost("/");  // 设置虚拟机路径
        // 一个 mq 的服务可以设置多个虚拟机路径，每个虚拟机路径对应一个虚拟机（相当于一个独立的 mq 服务），即每个虚拟机路径下的 mq 服务环境是相互隔离的。
        Connection connection = null;
        Channel channel = null;
        try {
            // 创建新连接
            connection = connectionFactory.newConnection();
            // 创建会话通道（channel），客户端和 mq 服务都是通过连接中的会话通道通信
            channel = connection.createChannel();
            // 声明队列，如果队列不存在，则创建
            /**
             * 参数列表：
             *  String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *      queue：队列名称
             *      durable：是否持久化，mq 重启后队列依旧存在
             *      exclusive：是否排他，独占该连接，队列仅能被此连接访问，如果连接关闭，则队列自动删除
             *      autoDelete：自动删除，队列不再使用时自动删除此队列，如果和 exclusive 都为 true，则定义的队列就是临时队列
             *      arguments：参数，可设置队列的扩展参数，如设置队列的存活时间等...
             */
            channel.queueDeclare(QUEUE_5, true, false, false, null);
            channel.queueDeclare(QUEUE_6, true, false, false, null);
            // 声明交换机
            /**
             * 参数列表：
             *  String exchange, String type
             *      exchange：交换机名称
             *      type：交换机类型，具体信息可参考上篇文章中的 Exchange 类型
             */
            channel.exchangeDeclare(TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC);

            // 绑定交换机和队列
            /**
             * 参数列表：
             *  String queue, String exchange, String routingKey
             *      queue：队列名称
             *      exchange：交换机名称
             *      routingKey：路由 key
             */
            channel.queueBind(QUEUE_5, TOPIC_EXCHANGE, QUEUE_5_ROUTING_KEY);
            channel.queueBind(QUEUE_6, TOPIC_EXCHANGE, QUEUE_6_ROUTING_KEY);

            // 发布消息给交换机
            /**
             * 参数列表：
             *  String exchange, String routingKey, BasicProperties props, byte[] body
             *      exchange：交换机，如果不指定则使用默认的交换机
             *      routingKey：路由键，交换机按路由键来将消息转发到指定的队列，如果使用默认交换机，那么 routingKey 要设为队列的名称
             *      props：消息属性
             *      body：消息体，消息内容
             */
            String routingKey = "";
            for (int i = 0; i < 10; i++) {
                String msg = "hello world" + i;
                routingKey = i % 2 == 0 ? "msg.queue_5" : "msg.queue_6";
                channel.basicPublish(TOPIC_EXCHANGE, routingKey, null, msg.getBytes());
                System.out.println("Send msg " + msg + " to " + TOPIC_EXCHANGE);
            }

            channel.basicPublish(TOPIC_EXCHANGE, "msg.queue_5.queue_6", null, "hello world to all queues".getBytes());

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) channel.close();
            if (connection != null) connection.close();
        }
    }
}

通配符模式对应的交换机类型为 Topic，所以该生产者程序在入门程序的基础上添加了手动定义的 Topic 类型的交换机，并且声明了 2 个队列绑定到了该交换机，绑定时的路由 key 分别为 msg.#.queue_5.# 和 msg.#.queue_6.#，之后分别发送了 5 条消息携带的路由 key 分别为 msg.queue_5 和 msg.queue_6，且在最后还发送了一条消息路由 key 为 msg.queue_5.queue_6 ，路由 key msg.queue_5 和 msg.queue_6 分别能匹配 msg.#.queue_5.# 和 msg.#.queue_6.# 绑定的队列，而 msg.queue_5.queue_6 既能匹配到 msg.#.queue_5.# 绑定的队列，也能匹配到 msg.#.queue_6.# 绑定的队列，所以两个队列都应该接收到 6 条消息。
执行该程序，查看会发现两个队列都接受到了消息：

Header 转发器模式

Header 模式与 Routing 模式类似，不同的地方在于 Header 模式不使用路由 key，而是使用 Header 中的 key/value（键值对）来匹配队列。

示例

import com.rabbitmq.client.*;

import java.util.HashMap;

/**
 * RabbitMQ Header 模式-生产者
 */
public class Producer05 {
    private static final String HEADER_EXCHANGE = "test_header_exchange";
    private static final String QUEUE_7 = "queue_7";
    private static final String QUEUE_8 = "queue_8";

    /**
     * queue_5 使用的路由 key
     */
    private static final String QUEUE_7_ROUTING_KEY = "queue_7_key";
    /**
     * queue_6 使用的路由 key
     */
    private static final String QUEUE_8_ROUTING_KEY = "queue_8_key";

    public static void main(String[] args) throws Exception {
        //  通过连接工厂创建和 mq 服务的连接
        ConnectionFactory connectionFactory = new ConnectionFactory();
        connectionFactory.setHost("192.168.146.131"); // mq 服务主机地址
        connectionFactory.setPort(5672);        // mq 服务暴露的端口
        connectionFactory.setUsername("guest"); // 用户名
        connectionFactory.setPassword("guest"); // 密码
        connectionFactory.setVirtualHost("/");  // 设置虚拟机路径
        // 一个 mq 的服务可以设置多个虚拟机路径，每个虚拟机路径对应一个虚拟机（相当于一个独立的 mq 服务），即每个虚拟机路径下的 mq 服务环境是相互隔离的。
        Connection connection = null;
        Channel channel = null;
        try {
            // 创建新连接
            connection = connectionFactory.newConnection();
            // 创建会话通道（channel），客户端和 mq 服务都是通过连接中的会话通道通信
            channel = connection.createChannel();
            // 声明队列，如果队列不存在，则创建
            /**
             * 参数列表：
             *  String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
             *      queue：队列名称
             *      durable：是否持久化，mq 重启后队列依旧存在
             *      exclusive：是否排他，独占该连接，队列仅能被此连接访问，如果连接关闭，则队列自动删除
             *      autoDelete：自动删除，队列不再使用时自动删除此队列，如果和 exclusive 都为 true，则定义的队列就是临时队列
             *      arguments：参数，可设置队列的扩展参数，如设置队列的存活时间等...
             */
            channel.queueDeclare(QUEUE_7, true, false, false, null);
            channel.queueDeclare(QUEUE_8, true, false, false, null);
            // 声明交换机
            /**
             * 参数列表：
             *  String exchange, String type
             *      exchange：交换机名称
             *      type：交换机类型，具体信息可参考上篇文章中的 Exchange 类型
             */
            channel.exchangeDeclare(HEADER_EXCHANGE, BuiltinExchangeType.HEADERS);

            // 绑定交换机和队列
            /**
             * 参数列表：
             *  String queue, String exchange, String routingKey, Map<String, Object> arguments
             *      queue：队列名称
             *      exchange：交换机名称
             *      routingKey：路由 key
             *      arguments：Headers 参数
             */
            HashMap<String, Object> queue7Map = new HashMap<>();
            queue7Map.put("routing_key",QUEUE_7_ROUTING_KEY);
            HashMap<String, Object> queue8Map = new HashMap<>();
            queue8Map.put("routing_key",QUEUE_8_ROUTING_KEY);

            channel.queueBind(QUEUE_7, HEADER_EXCHANGE, "",queue7Map);
            channel.queueBind(QUEUE_8, HEADER_EXCHANGE, "",queue8Map);

            // 发布消息给交换机
            /**
             * 参数列表：
             *  String exchange, String routingKey, BasicProperties props, byte[] body
             *      exchange：交换机，如果不指定则使用默认的交换机
             *      routingKey：路由键，交换机按路由键来将消息转发到指定的队列，如果使用默认交换机，那么 routingKey 要设为队列的名称
             *      props：消息属性
             *      body：消息体，消息内容
             */
            AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
            builder.headers(queue7Map);
            channel.basicPublish(HEADER_EXCHANGE, "", builder.build(), "hello queue_7".getBytes());
            builder.headers(queue8Map);
            channel.basicPublish(HEADER_EXCHANGE, "", builder.build(), "hello queue_8".getBytes());
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (channel != null) channel.close();
            if (connection != null) connection.close();
        }
    }
}

通配符模式对应的交换机类型为 Headers，所以该生产者程序在入门程序的基础上添加了手动定义的 Headers 类型的交换机，并且声明了 2 个队列绑定到了该交换机。
绑定时没有设置路由，而是设置了 map 属性，之后发送消息时也指定一个 map，交换机会根据 map 的键值匹配并将消息转发到对应的消息队列。
执行该程序，查看会发现两个队列分别接受到了 1 条消息：

RPC 远程过程调用模式

RPC 即客户端远程调用服务端的方法 ，使用 MQ 可以实现 RPC 的异步调用，基于 Direct 交换机实现，流程如下：

1. 客户端既是生产者就是消费者，向 RPC 请求队列发送 RPC 调用消息，同时监听 RPC 响应队列；
2. 服务端监听 RPC 请求队列的消息，收到消息后执行服务端的方法，得到方法返回的结果；
3. 服务端将 RPC 方法的结果发送到 RPC 响应队列；
4. 客户端（RPC 调用方）监听 RPC 响应队列，接收到 RPC 调用结果；

SpringBoot整合RabbitMQ

参考 https://www.zze.xyz/archives/springboot31.html