RabbitMQ 消息类型

RabbitMQ 消息类型

下面我们简单介绍下RabbitMQ的一些消息种类，并结合Java代码进行学习。

如果需要执行代码，需要下载RabbitMQ的客户端（例如java客户端： https://www.rabbitmq.com/java-client.html）
使用maven：

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

注意先启动消费者，在启动生产者
5.x 版本系列需要 JDK 8
java-client 的文档：https://rabbitmq.github.io/rabbitmq-java-client/api/current/index.html

创建一个连接工具类ConnectionUtil

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

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

/**
 * 类描述：
 * 连接工具类
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 13:49
 */
public class ConnectionUtil {

    /**
     * 获取MQ的连接
     * @return
     */
    public static Connection getConnection() throws IOException, TimeoutException {
        // 定义一个连接工厂
        ConnectionFactory factory = new ConnectionFactory();
        // 设置服务地址
        factory.setHost("localhost");
        // AMQP 5672
        factory.setPort(5672);
        // vhost
        factory.setVirtualHost("/vhost01");
        // 用户名
        factory.setUsername("admin");
        // 密码
        factory.setPassword("123456");
        return factory.newConnection();
    }
}

simple 简单队列

​P:消息生产者
​红色：队列
​C：消费者
​3个对象：生产者 队列 消费者

生产者直接发送消息到队列，消费者直接从队列获取消息。发送消息时，只需要指定队列，不需要指定交换机，以及路由key，只有一个消费者。

示例

1. 消息生产者：Send

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * 类描述：
 * 简单队列，消息生产者
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 13:58
 */
public class Send {
    private static final String QUEUE_NAME="test_simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取一个连接
        Connection connection = ConnectionUtil.getConnection();

        // 从连接中获取一个通道
        Channel channel = connection.createChannel();

        // 声明一个队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        String message = "hello simple !";

        // 第一个参数是exchangeName(默认情况下代理服务器端是存在一个""名字的exchange的,
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());

        System.out.println("send message");

        channel.close();
        connection.close();
    }
}

2. 消息消费者：Recv

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 简单队列，消息消费者
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:00
 */
public class Recv {
    private static final String QUEUE_NAME="test_simple_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();

        // 创建通道
        Channel channel = connection.createChannel();

        // 队列声明
        channel.queueDeclare(QUEUE_NAME, false, false,false, null);

        // 定义队列的消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
            // 获取到达的消息
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msgString = new String(body,"utf-8");
                System.out.println("new api recv: " + msgString);
            }
        };

        // 监听队列
        channel.basicConsume(QUEUE_NAME, true, defaultConsumer);
    }
}

简单队列的不足：耦合性高，生产者一一对应消费者（如果我想有多个消费者消费队列中消息，这时候就不行），队列名变更，这时候得同时变更。

work queues 工作队列

工作队列可以细分为轮询分发和公平分发。发送消息时，只需要指定队列，不需要指定交换机，以及路由key，设定多个消费者。

为什么会出现工作队列，因为simple 队列是一一对应的，而且我们实际开发，生产者发送消息是毫不费力的，而消费者一般是要跟业务相结合的，消费者接收到消息之后就需要处理，可能需要花费时间，这时候队列就会积压了很多消息。

round robin 轮询分发

1. 消息生产者：Send

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * 类描述：
 * 工作队列，轮询分发，消息生产者
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:12
 */
public class Send {
    public static final String QUEUE_NAME = "test_round_robin_work_queue";

    /**
     *                  |--> C2
     * P ---> Queue ----|
     *                  |--> C1
     * @param args
     * @throws IOException
     * @throws TimeoutException
     */
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        // 发送消息
        for (int i = 0; i < 50; i++) {
            String msg= "send hello " + i;
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.println("【WQ】 send msg = " + msg);
            Thread.sleep(i*20);
        }
        // 关闭资源
        channel.close();
        connection.close();
    }
}

2. 消息消费者：Recv1

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 工作队列，轮询分发，消息消费者1号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:14
 */
public class Recv1 {
    public static final String QUEUE_NAME = "test_round_robin_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        // 定义一个消费者
        Consumer consumer = new DefaultConsumer(channel) {
            // 消息到达，触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [1] msg = " + msg);
                try{
                    Thread.sleep(2000);
                }catch (InterruptedException e) {
                    e.printStackTrace();
                }finally{
                    System.out.println("Recv [1] done!");
                }
            }
        };

        // 监听队列
        boolean autoAck = true;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

3. 消息消费者：Recv2

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 工作队列，轮询分发，消息消费者2号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:14
 */
public class Recv2 {
    public static final String QUEUE_NAME = "test_round_robin_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        // 定义一个消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            // 消息到达，触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [2] msg = " + msg);
                try{
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally{
                    System.out.println("Recv [2] done!");
                }
            }
        };

        // 监听队列
        boolean autoAck = true;
        channel.basicConsume(QUEUE_NAME, autoAck, defaultConsumer);
    }
}

现象：

• 消费者1 和消费者2处理消息的数量是一样的。
• 消费者1：偶数。
• 消费者2：奇数。
• 这种方式叫做轮询分发（round-robin）结果就是不管谁忙或者谁清闲 都不会多给一个消息，任务总是你一个我一个。

fair dispatch 公平分发

公平分发，需要消费者进行手动回执

// MQ一次只发一个请求给消费者，当消费者处理完消息后会手动回执，然后MQ再发一个消息给消费者
channel.basicQos(1);

boolean autoAck = false; //false 手动回执,处理完消息后，告诉MQ
channel.basicConsume(QUEUE_NAME, autoAck, defaultConsumer);

1. 消息生产者：Send

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * 类描述：
 * 工作队列，公平分发，消息生产者
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:30
 */
public class Send {

    public static final String QUEUE_NAME = "test_fair_dispatch_work_queue";

    /**
     *                  |--> C2
     * P ---> Queue ----|
     *                  |--> C1
     * @param args
     * @throws IOException
     * @throws TimeoutException
     */
    public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false, null);

        /**
         * 每个消费者 发送确认消息之前，消息队列不发送下一个消息到消费者，一次只处理一个消息
         * 限制发送给同一个消费者 不得超过一条消息
         */
        channel.basicQos(1);

        // 发送消息
        for (int i = 0; i < 50; i++) {
            String msg= "send hello " + i;
            channel.basicPublish("", QUEUE_NAME, null, msg.getBytes());
            System.out.println("【WQ】 send msg = " + msg);
            Thread.sleep(i*5);
        }
        // 关闭资源
        channel.close();
        connection.close();
    }
}

2. 消息消费者：Recv1

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 工作队列，公平分发，消息消费者1号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:14
 */
public class Recv1 {
    public static final String QUEUE_NAME = "test_fair_dispatch_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        channel.basicQos(1);

        // 定义一个消费者
        Consumer consumer = new DefaultConsumer(channel) {
            // 消息到达，触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [1] msg = " + msg);
                try{
                    Thread.sleep(2000);
                }catch (InterruptedException e) {
                    e.printStackTrace();
                }finally{
                    System.out.println("Recv [1] done!");

                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        // 监听队列
        // boolean autoAck = true; //自动应答
        boolean autoAck = false; //手动应答
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

3. 消息消费者：Recv2

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 工作队列，公平分发，消息消费者2号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:14
 */
public class Recv2 {
    public static final String QUEUE_NAME = "test_fair_dispatch_work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        channel.basicQos(1);
        // 定义一个消费者
        DefaultConsumer defaultConsumer = new DefaultConsumer(channel) {
            // 消息到达，触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [2] msg = " + msg);
                try{
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally{
                    System.out.println("Recv [2] done!");
                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        // 监听队列
        boolean autoAck = false; //false 手动回执
        channel.basicConsume(QUEUE_NAME, autoAck, defaultConsumer);
    }
}

现象：消费者2 处理的消息比消费者1多，能者多劳

publish/subscribe 发布-订阅模式

• 一个生产者，多个消费者，需要新建fanout交换机
• 每个消费者都有自己的队列,并绑定到交换机上
• 生产者没有直接把消息发送到队列，而是发送到交换机
• 消息发送时需要指定交换机，消息接收时需要指定队列
• 生产者发送的消息，经过交换机，到达队列。就能实现一个消息被多个消费者消费

注册成功时，既要发邮件，又要发短信

示例

1. 消息生产者：Send

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * 类描述：
 * 发布订阅模式队列，消息生产者
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:51
 */
public class Send {

    public static final String EXCHANGE_NAME = "test_exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        // 声明交换机
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout"); //分发

        // 发送消息
        String msg = "hello ps";
        channel.basicPublish(EXCHANGE_NAME, "", null, msg.getBytes());
        System.out.println("Send msg = " + msg);
        channel.close();
        connection.close();
    }
}

消息哪去了？？ 丢失了，因为交换机没有存储的能力，在rabbitmq里面只有队列有存储的能力。因为还没有队列绑定到这个交换机，所以数据丢失了。

2. 消息消费者：Recv1

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 发布订阅模式队列，消息消费者1号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:59
 */
public class Recv1 {

    public static final String QUEUE_NAME = "test_queue_fanout_email";

    public static final String EXCHANGE_NAME = "test_exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        // 保证一次只分发一个
        channel.basicQos(1);

        // 绑定到交换机 转发器
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");

        // 定义一个消费者
        Consumer consumer = new DefaultConsumer(channel) {
            // 消息到达，触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [1] msg = " + msg);
                try{
                    Thread.sleep(2000);
                }catch (InterruptedException e) {
                    e.printStackTrace();
                }finally{
                    System.out.println("Recv [1] done!");
                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        // 监听队列
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

3. 消息消费者：Recv2

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

/**
 * 类描述：
 * 发布订阅模式队列，消息消费者1号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 14:59
 */
public class Recv2 {

    public static final String QUEUE_NAME = "test_queue_fanout_sms";

    public static final String EXCHANGE_NAME = "test_exchange_fanout";

    public static void main(String[] args) throws IOException, TimeoutException {
        // 获取连接
        Connection connection = ConnectionUtil.getConnection();
        // 获取channel
        Channel channel = connection.createChannel();
        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        // 绑定到交换机 转发器
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "");
        // 保证一次只分发一个
        channel.basicQos(1);
        // 定义一个消费者
        Consumer consumer = new DefaultConsumer(channel) {
            // 消息到达，触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [2] msg = " + msg);
                try{
                    Thread.sleep(2000);
                }catch (InterruptedException e) {
                    e.printStackTrace();
                }finally{
                    System.out.println("Recv [2] done!");
                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        // 监听队列
        boolean autoAck = false;
        channel.basicConsume(QUEUE_NAME, autoAck, consumer);
    }
}

现象：消费者1和消费者2都受到了消息。

routing 路由选择通配符模式

• direct交换机类型
• 生产者需要将交换机和routing key绑定
• 消费者需要将队列，交换机，routing key 三者绑定
• 每个消息会根据不同的Routing key，发送到不同的消费者队列
• 不支持通配符（*,#）

示例

1. 消息生产者：Send

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * 类描述：
 * 路由模式队列，消息生产者
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 15:12
 */
public class Send {
    public static final String EXCHANGE_NAME = "test_exchange_direct";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        // exchange direct:直连
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");

        String msg = "hello direct !";
        // routing key
        String routingKey = "info";
        channel.basicPublish(EXCHANGE_NAME, routingKey, null, msg.getBytes());

        System.out.println("send :" + msg);

        channel.close();
        connection.close();
    }
}

2. 消息消费者：Recv1

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 路由模式队列，消息消费者1号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 15:13
 */
public class Recv1 {
    public static final String EXCHANGE_NAME = "test_exchange_direct";

    public static final String QUEUE_NAME = "test_queue_direct_1";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        // 绑定队列到交换机，并绑定 routingKey
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");

        // 保证一次只分发一个
        channel.basicQos(1);

        // 定义一个消费者
        Consumer consumer = new DefaultConsumer(channel){
            // 消息到达，触发这个方法
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [1] msg = " + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally{
                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                    System.out.println("Recv [1] done!");
                }
            }
        };

        // 监听队列 autoAck(消息应答):false 手动回执 (消息回执 channel.basicAck(envelope.getDeliveryTag(), false);)
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

3. 消息消费者：Recv2

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 路由模式队列，消息消费者2号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 15:13
 */
public class Recv2 {
    public static final String EXCHANGE_NAME = "test_exchange_direct";

    public static final String QUEUE_NAME = "test_queue_direct_2";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false,null);

        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "error");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "info");
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "waring");

        channel.basicQos(1);

        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("Recv [2] msg = " + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally{
                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                    System.out.println("Recv [2] done!");
                }
            }
        };

        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

现象：队列绑定的路由key和消息发送时指定的路由key匹配时，才会接收到消息。

Topics 主题

• Topic交换机类型,生产者需要申明topic交换机,并指定routing key。
• ​和路由模式类似，但是Topic可以支持通配符，# 匹配一个或者多个字符；​* 匹配一个字符

商品：发布、删除、修改、查询

1. 消息生产者：Send

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;

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

/**
 * 类描述：
 * 主题队列，消息生产者
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 15:21
 */
public class Send {
    public static final String EXCHANGE_NAME = "test_exchange_topic";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        // 声明交换机 topic：主题模式
        channel.exchangeDeclare(EXCHANGE_NAME, "topic");

        String message = "商品...";

        // goods.delete 只有消费者2号能收到（goods.#）
        channel.basicPublish(EXCHANGE_NAME, "goods.delete", null, message.getBytes());
        // 两个消费者都能收到
        // channel.basicPublish(EXCHANGE_NAME, "goods.add", null, message.getBytes());

        System.out.println("send message = " + message);

        channel.close();
        connection.close();
    }
}

2. 消息消费者：Recv1

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 主题队列，消息消费者1号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 15:21
 */
public class Recv1 {
    public static final String EXCHANGE_NAME = "test_exchange_topic";

    public static final String QUEUE_NAME = "test_queue_topic_1";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false,false,false,null);

        //绑定 商品新增 goods.add
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "goods.add");

        channel.basicQos(1);

        //定义消费者
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("[1] recv msg: " + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally{
                    System.out.println("[1] recv done!");
                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        // 监听队列
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

3. 消息消费者：Recv2

import com.goudong.modules.rabbitmq.demo.util.ConnectionUtil;
import com.rabbitmq.client.*;

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

/**
 * 类描述：
 * 主题队列，消息消费者2号
 * @author cfl
 * @version 1.0
 * @date 2022/10/10 15:21
 */
public class Recv2 {
    public static final String EXCHANGE_NAME = "test_exchange_topic";

    public static final String QUEUE_NAME = "test_queue_topic_2";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtil.getConnection();

        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false,false,false,null);

        //绑定 goods.# (`#` 匹配一个或者多个字符)
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "goods.#");

        channel.basicQos(1);

        //定义消费者
        Consumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String msg = new String(body, "utf-8");
                System.out.println("[2] recv msg: " + msg);
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally{
                    System.out.println("[2] recv done!");
                    // 手动回执 false 表示只确认 envelope.DelivertTag 这条消息，true 表示确认 小于等于 b.DelivertTag 的所有消息（批量确认）
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };

        // 监听队列
        channel.basicConsume(QUEUE_NAME, false, consumer);
    }
}

现象：只有满足生产者指定的路由模式，才会将消息发送到队列。