• RabbitMQ官方案例学习记录

    官方文档：RabbitMQ教程 — RabbitMQ (rabbitmq.com)

    一、安装RabbitMQ服务

    直接使用docker在服务器上安装

    docker run -it -d --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.12-management

    安装完成后，访问15672端口，默认用户名和密码都是 guest，即可进入 

    二、hello world——梦开始的地方

    1. 介绍

            RabbitMQ 是一个消息代理：它接受并转发消息。 你可以把它想象成一个邮局：当你把你想要邮寄的邮件放在邮箱里时， 您可以确定，邮递员最终会将邮件递送给您的收件人。 在这个类比中，RabbitMQ是一个邮政信箱，一个邮局和一个信件载体。RabbitMQ和邮局的主要区别在于它不处理纸张， 相反，它接受、存储和转发数据的二进制 blob - 消息。

    2. 一些术语

    生产者：消息的发送方

    队列queue：本质上是一个大型的消息缓冲区

    消费者：消息的使用方

    Channel 频道：理解为操作消息队列的 client（比如 jdbcClient、redisClient），提供了和消息队列 server 建立通信的传输方法（为了复用连接，提高传输效率）。程序通过 channel 操作 rabbitmq（收发消息）

    3. 编写代码

    用 Java 编写两个程序;一个发送单个消息的生产者，一个接收的使用者并将消息打印出来。

    （1）消息生产者：

    编码过程：

    先创建连接工厂，然后通过工厂创建连接，再通过连接创建channel。通过channel来绑定队列或者交换机，再用channel来生产或者消费消息。

    channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());

    对于这行代码，可以看到消息是根据QUEUE_NAME路由到对应的队列。

    1. import com.rabbitmq.client.ConnectionFactory;
    2. import com.rabbitmq.client.Connection;
    3. import com.rabbitmq.client.Channel;
    4. public class MQProducer {
    5.  
    6.     //设置队列名
    7.     private final static String QUEUE_NAME = "hello-zy";
    8.     public static void main(String[] argv) throws Exception {
    9.         //创建连接工厂
    10.         ConnectionFactory factory = new ConnectionFactory();
    11.         //设置连接的服务器IP和端口号
    12.         factory.setHost("123.249.112.12");
    13.         factory.setPort(5672);
    14.         //创建一个连接和通道，这里使用 try-with-resources语句，
    15.         // 因为Connection和Channel都实现了java.lang.AutoCloseable，不需要在代码中再显式的关闭他们
    16.         try(Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
    17.  
    18.         /*
    19.         创建队列 channel.queueDeclare()用于声明一个队列
    20.         queue（队列名称）：指定要声明的队列的名称。
    21.         durable（持久化）：指定队列是否是持久化的。当 RabbitMQ 重新启动时，持久化的队列将被保留下来。如果将该参数设置为 true，则队列将被持久化；如果设置为 false，则队列不会被持久化。注意，这里指的是队列本身的持久化，而不是队列中的消息。
    22.         exclusive（排他性）：指定队列是否是排他的。如果将该参数设置为 true，则该队列只能被当前连接的消费者使用，并且在连接关闭时会自动删除该队列。如果设置为 false，则队列可供多个消费者使用。
    23.         autoDelete（自动删除）：指定队列在不再被使用时是否自动删除。如果将该参数设置为 true，则当队列不再被消费者使用时，将自动删除该队列。如果设置为 false，则队列不会自动删除。
    24.         arguments（参数）：指定队列的其他属性和参数，以键值对的形式提供。
    25.         */
    26.             channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    27.             //发送消息
    28.             String msg = "hello,world! RabbitMQ!";
    29.             //channel.basicPublish("", QUEUE_NAME, null, msg.getBytes())
    30.             // 这行代码的作用是将 msg 消息发布到默认交换器（空字符串）并使用QUEUE_NAME作为路由键，
    31.             // 消息的属性设置为默认值，消息的内容为 msg.getBytes()，即将 msg 转换为字节数组后发送。
    32.             channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
    33.             System.out.println(" 生产者发送消息：'" + msg + "'");
    34.         }
    35.     }
    36. }

     然后运行这个生产者代码，去网页端查看：

    （2）消息消费者：

    1. import com.rabbitmq.client.Channel;
    2. import com.rabbitmq.client.Connection;
    3. import com.rabbitmq.client.ConnectionFactory;
    4. import com.rabbitmq.client.DeliverCallback;
    5.  
    6. import java.io.IOException;
    7. import java.util.concurrent.TimeoutException;
    8.  
    9. public class MQConsumer {
    10.  
    11.     //声明队列 和消息发送方保持一致
    12.     private final static String QUEUE_NAME = "hello-zy";
    13.  
    14.     public static void main(String[] args) throws IOException, TimeoutException {
    15.         ConnectionFactory factory = new ConnectionFactory();
    16.         //设置rabbitMQ服务端ip
    17.         factory.setHost("123.249.112.12");
    18.         //这里不用try-with-resources 因为消费方需要一致保持监听，不要关闭
    19.         Connection connection = factory.newConnection();
    20.         Channel channel = connection.createChannel();
    21.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    22.         System.out.println(" 等待接收消息，退出请按 CTRL+C");
    23.  
    24.         //创建了一个消费者对象并实现了 handleDelivery() 方法作为回调方法。当消费者收到消息时，将自动执行该方法。
    25.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    26.             String message = new String(delivery.getBody(), "UTF-8");
    27.             System.out.println(" 消费了消息：'" + message + "'");
    28.         };
    29.         channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    30.     }
    31. }

    去网页端查看消息是否被消费，可以看到多了一个消费者，并且消息已经没了。 

    三、工作队列 WorkQueue

            工作队列（又名：任务队列）背后的主要思想是避免立即执行资源密集型任务，必须等待它要完成。相反，我们将任务安排在以后完成。我们将任务封装为消息并将其发送到队列。正在运行的工作进程 在后台将弹出任务并最终执行工作。当您运行许多工作线程时，任务将在它们之间共享。

    WorkQueue的模型跟前面第一个案例Hello，World！的模型，最明显的区别其实就是，第一个案例他只有一个消费者。我们知道RabbitMQ他的消息是阅完即焚，即消费者一旦接收，这个消息直接就从Queue中被弹出了。
    而现在这个案例，他有两个消费者（画两个只是方便，他当然也可以有3个、4个），他的消息应该是通过某种算法做负载均衡送到不同的消费者，让消费者进行处理，让消息不至于处理不过来，从而导致滞留在Queue中的消息被弹出。

    思路如下：
    1、我们先让Publish服务每秒发布50条消息到 simple.queue，来演示消息的频繁发送。
    2、在Consumer服务中定义两个消费者，来监听我们的 simple.queue队列。
    3、消费者1每秒处理40条消息，消费者2每秒处理30条消息。

    1. 循环调度

    （1）生产者：

    生成50条消息：

    1. public class Send {
    2.  
    3.     //设置队列名
    4.     private final static String QUEUE_NAME = "hello-zy";
    5.  
    6.     public static void main(String[] argv) throws Exception {
    7.         //创建连接
    8.         ConnectionFactory factory = new ConnectionFactory();
    9.         //设置连接的服务器IP和端口号
    10.         factory.setHost("123.249.112.12");
    11.         factory.setPort(5672);
    12.         //创建一个通道，这里使用 try-with-resources语句，
    13.         // 因为Connection和Channel都实现了java.lang.AutoCloseable，不需要在代码中再显式的关闭他们
    14.         try(Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
    15.  
    16.             channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    17.             //发送消息
    18.  
    19.             for(int i = 1; i <= 50; i++) {
    20.                 String msg = "hello, I am";
    21.                 msg = msg + i;
    22.                 channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
    23.                 Thread.sleep(200);
    24.                 System.out.println(" 生产者发送消息：'" + msg + "'");
    25.             }
    26.         }
    27.     }
    28. }

    （2）消费者1和消费者2

    1. public class Worker_01 {
    2.  
    3.     //声明队列 和消息发送方保持一致
    4.     private final static String QUEUE_NAME = "hello-zy";
    5.  
    6.     public static void main(String[] args) throws IOException, TimeoutException {
    7.         ConnectionFactory factory = new ConnectionFactory();
    8.         //设置rabbitMQ服务端ip
    9.         factory.setHost("123.249.112.12");
    10.         //这里不用try-with-resources 因为消费方需要一致保持监听，不要关闭
    11.         Connection connection = factory.newConnection();
    12.         Channel channel = connection.createChannel();
    13.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    14.         System.out.println(" 等待接收消息，退出请按 CTRL+C");
    15.  
    16.         //创建了一个消费者对象并实现了 handleDelivery() 方法作为回调方法。当消费者收到消息时，将自动执行该方法。
    17.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    18.             String message = new String(delivery.getBody(), "UTF-8");
    19.             System.out.println(" Worker_01消费了消息：'" + message + "'");
    20.         };
    21.         channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    22.     }
    23. }
    24.  
    25. public class Worker_02 {
    26.  
    27.     //声明队列 和消息发送方保持一致
    28.     private final static String QUEUE_NAME = "hello-zy";
    29.  
    30.     public static void main(String[] args) throws IOException, TimeoutException {
    31.         ConnectionFactory factory = new ConnectionFactory();
    32.         //设置rabbitMQ服务端ip
    33.         factory.setHost("123.249.112.12");
    34.         //这里不用try-with-resources 因为消费方需要一致保持监听，不要关闭
    35.         Connection connection = factory.newConnection();
    36.         Channel channel = connection.createChannel();
    37.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    38.         System.out.println(" 等待接收消息，退出请按 CTRL+C");
    39.  
    40.         //创建了一个消费者对象并实现了 handleDelivery() 方法作为回调方法。当消费者收到消息时，将自动执行该方法。
    41.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    42.             String message = new String(delivery.getBody(), "UTF-8");
    43.             System.out.println(" Worker_02消费了消息：'" + message + "'");
    44.         };
    45.         channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
    46.     }
    47. }

    启动两个消费者，然后开始生产消息，控制台打印如下：

     可以看到是交替消费的：

    2. 消息确认

            在上面的代码中，一旦消息传递给消费者，就会立即被删除。如果在消费过程中，消费者宕机，消息没消费成功，但是因为已经投递出去了，消息从队列删掉。就会出现：消息未消费，且丢失的问题。对于这种情况，我们希望没消费成功的消息，转交给其他的消费者消费。

            为了确保消息永远不会丢失，RabbitMQ 支持消息确认。确认由消费者告诉 RabbitMQ 已收到特定消息，并处理，RabbitMQ可以自由删除它。

            如果消费者没有发送确认消息，rabbitMQ可以知道消息没有消费成功，并将重新排队。

            在消费者返回确认消息时强制实施超时（默认为 30 分钟）。 这有助于检测从不确认交付的错误（卡住）消费者。 可以按照传递确认超时中所述增加此超时。

            默认情况下，手动消息确认处于打开状态。在上一个 示例，我们通过 autoAck=true 标志明确关闭了它们。是时候将此标志设置为 false ，来使消费者发送确认消息。

    （1）消费者

    生产者还是不变，生成10条消息到队列中：

    创建消费者，启动手动确认：

    1. public class Consumer {
    2.     private final static String QUEUE_NAME = "hello-zy";
    3.  
    4.     public static void main(String[] argv) throws Exception {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("123.249.112.12");
    7.         factory.setPort(5672);
    8.  
    9.         Connection connection = factory.newConnection();
    10.         Channel channel = connection.createChannel();
    11.             
    12.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    13.             
    14.             // 将autoAck参数设置为false，关闭自动消息确认
    15.             /*
    16.             * new DefaultConsumer(channel) { ... }：这是一个匿名内部类，用于定义消息处理的逻辑。
    17.             * 它继承自DefaultConsumer，并覆盖了 handleDelivery 方法，以自定义消息的处理方式。
    18.             * */
    19.         channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {
    20.  
    21.             /*handleDelivery 方法：这是 DefaultConsumer 类中的方法，用于处理从队列中接收的消息。
    22.               consumerTag：标识消费者的标签。
    23.               envelope：包含与消息相关的元数据，如交付标签、交付模式等。
    24.               properties：包含消息的属性，如消息的头部信息。
    25.               body：消息的内容，以字节数组形式提供。
    26.             */
    27.             @Override
    28.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    29.                 String message = new String(body, "UTF-8");
    30.                 System.out.println("消费者接收消息: '" + message + "'");
    31.                     
    32.                 // 在消息处理成功后，发送确认消息
    33.                 channel.basicAck(envelope.getDeliveryTag(), false);
    34.                 System.out.println("确认收到了消息："+message);
    35.             }
    36.         });
    37.     }
    38. }

    启动消费者：

    （2）验证未成功消费情况

    先生产6条消息到队列：

    简单修改一下消费者代码，在中间添加判断逻辑，当碰到消息5的时候，退出消费者：

    1. public class Consumer1 {
    2.     private final static String QUEUE_NAME = "hello-zy";
    3.  
    4.     public static void main(String[] argv) throws Exception {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("123.249.112.12");
    7.  
    8.         Connection connection = factory.newConnection();
    9.         Channel channel = connection.createChannel();
    10.             
    11.         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    12.  
    13.         try {
    14.             // 将autoAck参数设置为false，关闭自动消息确认
    15.             channel.basicConsume(QUEUE_NAME, false, new DefaultConsumer(channel) {
    16.                 @Override
    17.                 public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
    18.                         throws IOException {
    19.                     String message = new String(body, "UTF-8");
    20.                     System.out.println("消费者接收消息: '" + message + "'");
    21.                     
    22.                     // 模拟某个条件，例如消息处理成功
    23.                     if (!message.contains("5")) {
    24.                         // 在消息处理成功后，发送确认消息
    25.                         channel.basicAck(envelope.getDeliveryTag(), false);
    26.                         System.out.println("确认收到了消息：" + message);
    27.                     }
    28.                     
    29.                     // 模拟消费者退出
    30.                     if (message.contains("5")) {
    31.                         throw new RuntimeException("Consumer exiting...");
    32.                     }
    33.                 }
    34.             });
    35.         } catch (RuntimeException e) {
    36.             // 当消费者退出时，捕获异常
    37.             System.out.println("Consumer exited.");
    38.         }
    39.     }
    40. }

     

    可以看到，当消费者消费到第五条消息后，因为抛出了异常，所以后面的消息都未消费成功，所以会把第五条消息和第六条消息再放回消息队列，查看消息队列，可以看到还有两条消息在队列中：

    如果把参数改成true，也就是自动确认：这就会导致未消费的消息丢失

     

    3. 消息持久性

    我们已经保证了消息未消费的情况下不会丢失，但是如果RabbitMQ服务器宕机，消息还是会丢失。这就涉及到一个消息持久化的问题。

    需要做两件事来确保 消息不会丢失：我们需要将队列和消息都标记为durable。 （重新定义一个队列）

    此时，我们确信durable_queue队列不会丢失 即使 RabbitMQ 重新启动。现在我们需要将我们的消息标记为持久 - 通过设置消息属性（实现基本属性） 到值PERSISTENT_TEXT_PLAIN。 

     

    如果不设置未持久性，重启docker的rabbitmq容器，队列消息就丢了，设置durable=true后， 即使重启docker容器，队列和消息都不会丢失。

    四、发布/订阅模式

            在前面，都是一条消息由一个消费者消费。如果一条消息需要被多个消费者消费，那么就需要引入发布/订阅模式。

    1. 交换机

    关于交换机的概念（）：

    一个生产者给 多个 队列发消息，1 个生产者对多个队列。
    交换机的作用：提供消息转发功能，类似于网络路由器
    要解决的问题：怎么把消息转发到不同的队列上，好让消费者从不同的队列消费。

    交换机有多种类别：fanout、direct, topic, headers

    fanout（扇出）

    扇出、广播
    特点：消息会被转发到所有绑定到该交换机的队列
    场景：很适用于发布订阅的场景。比如写日志，可以多个系统间共享

    Direct 直接

    绑定：可以让交换机和队列进行关联，可以指定让交互机把什么样的消息发送给哪个队列（类似于计算机网络中，两个路由器，或者网络设备相互连接，也可以理解为网线）
    routingKey：路由键，控制消息要转发给哪个队列的（IP 地址） 

    特点：消息会根据路由键转发到指定的队列
    场景：特定的消息只交给特定的系统（程序）来处理
    绑定关系：完全匹配字符串

    比如发日志的场景，希望用独立的程序来处理不同级别的日志，比如 C1 系统处理 error 日志，C2 系统处理其他级别的日志

     

     topic 交换机

    特点：消息会根据一个 模糊的 路由键转发到指定的队列
    场景：特定的一类消息可以交给特定的一类系统（程序）来处理
    绑定关系：可以模糊匹配多个绑定
    ●*：匹配一个单词，比如 *.orange，那么 a.orange、b.orange 都能匹配
    ●#：匹配 0 个或多个单词，比如 a.#，那么 a.a、a.b、a.a.a 都能匹配

    注意，这里的匹配和 MySQL 的like 的 % 不一样，只能按照单词来匹配，每个 '.' 分隔单词，如果是 '#.'，其实可以忽略，匹配 0 个词也 ok

    Headers 交换机 

    类似主题和直接交换机，可以根据 headers 中的内容来指定发送到哪个队列。使用消息头headers来路由消息。

     2. 使用fanout交换机来实现发布/订阅

    （1）生产者

    1. public class LogProducer {
    2.     private final static String EXCHANGE_NAME = "logs";
    3.  
    4.     public static void main(String[] argv) throws Exception {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("123.249.112.12");
    7.         factory.setPort(5672);
    8.  
    9.         try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) {
    10.             //声明交换机类型为：FANOUT
    11.             channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    12.  
    13.             for (int i = 1; i <= 5; i++) {
    14.                 String message = "Log message " + i;
    15.                 //消息发到交换机中，而不是像之前点对点那样直接发到消息队列中
    16.                 channel.basicPublish(EXCHANGE_NAME, "", MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    17.                 System.out.println("生产者发送日志: " + message);
    18.             }
    19.         }
    20.     }
    21. }

    查看交换机列表，可以看到新增了一个名为logs的交换机，类型为 fanout：

     （2）消费者

    创建两个消费者，都绑定一个交换机，名字为log

    1. public class LogConsumer {
    2.     private final static String EXCHANGE_NAME = "logs";
    3.  
    4.     public static void main(String[] argv) throws Exception {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("123.249.112.12");
    7.  
    8.  
    9.         Connection connection = factory.newConnection();
    10.         Channel channel = connection.createChannel();
    11.  
    12.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    13.         
    14.         // 创建一个临时队列
    15.         String queueName = channel.queueDeclare().getQueue();
    16.         
    17.         // 将队列绑定到交换器
    18.         channel.queueBind(queueName, EXCHANGE_NAME, "");
    19.  
    20.         System.out.println("等待接收日志消息...");
    21.  
    22.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    23.             String message = new String(delivery.getBody(), "UTF-8");
    24.             System.out.println("消费者1接收消息: '" + message + "'");
    25.         };
    26.  
    27.         channel.basicConsume(queueName, true, deliverCallback, consumerTag -> { });
    28.     }
    29. }
    30. public class LogConsumer2 {
    31.     private final static String EXCHANGE_NAME = "logs";
    32.  
    33.     public static void main(String[] argv) throws Exception {
    34.         ConnectionFactory factory = new ConnectionFactory();
    35.         factory.setHost("123.249.112.12");
    36.  
    37.  
    38.         Connection connection = factory.newConnection();
    39.         Channel channel = connection.createChannel();
    40.  
    41.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
    42.         
    43.         // 创建一个临时队列
    44.         String queueName = channel.queueDeclare().getQueue();
    45.         
    46.         // 将队列绑定到交换器
    47.         channel.queueBind(queueName, EXCHANGE_NAME, "");
    48.  
    49.         System.out.println("等待接收日志消息...");
    50.  
    51.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    52.             String message = new String(delivery.getBody(), "UTF-8");
    53.             System.out.println("消费者2接收消息: '" + message + "'");
    54.         };
    55.  
    56.         channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
    57.         });
    58.     }
    59. }

     启动两个消费者，然后启动生产者发送消息，可以看到两个消费者都消费了消息：

    五、路由

    在上面的例子中，我们实现了一个简单的日志记录系统。能够向多个消费者广播消息。

    下面要增加一个新的功能：

            例如，我们将只能将关键错误消息定向到 日志文件（以节省磁盘空间），同时仍然能够打印所有控制台上的日志消息。

    1. 绑定队列和交换机

    之前是这样绑定的：第三个参数就是路由键

    channel.queueBind(queueName, EXCHANGE_NAME, "");

    绑定是交换和队列之间的关系。

            绑定可以采用额外的路由键参数。为了避免 与basic_publish参数混淆，我们将它称为绑定键。绑定键的含义取决于交换类型。

    channel.queueBind(queueName, EXCHANGE_NAME, "black");

    2. 使用direct交换机绑定

    （1）生产者

    模拟三条不同的消息，指定消息1的路由键为orange，消息2的路由键为black，消息3的路由键为green。

    1. public class LogProducer {
    2.     private final static String EXCHANGE_NAME = "direct_logs";
    3.  
    4.     public static void main(String[] argv) throws Exception {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("123.249.112.12");
    7.         factory.setPort(5672);
    8.  
    9.         try (Connection connection = factory.newConnection();
    10.              Channel channel = connection.createChannel()) {
    11.             //声明交换机类型为 direct
    12.             channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    13.  
    14.             String message1 = "Message with routing key orange";
    15.             String message2 = "Message with routing key black";
    16.             String message3 = "Message with routing key green";
    17.  
    18.             // 发布消息到交换器，并指定不同的路由键
    19.             channel.basicPublish(EXCHANGE_NAME, "orange", null, message1.getBytes());
    20.             channel.basicPublish(EXCHANGE_NAME, "black", null, message2.getBytes());
    21.             channel.basicPublish(EXCHANGE_NAME, "green", null, message3.getBytes());
    22.  
    23.             System.out.println("生产者发送消息完成.");
    24.         }
    25.     }
    26. }

    （2）消费者

    创建两个消费者

    消费者1和消费者2都绑定对应的交换机，其中消费者1对应路由键orange，消费者2对应路由键black和orange。

    1. public class LogConsumer1 {
    2.     private final static String EXCHANGE_NAME = "direct_logs";
    3.  
    4.     public static void main(String[] argv) throws Exception {
    5.         ConnectionFactory factory = new ConnectionFactory();
    6.         factory.setHost("123.249.112.12");
    7.         factory.setPort(5672);
    8.  
    9.         Connection connection = factory.newConnection();
    10.         Channel channel = connection.createChannel();
    11.  
    12.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    13.         
    14.         // 创建一个临时队列
    15.         String queueName = channel.queueDeclare().getQueue();
    16.         
    17.         // 绑定队列到交换器，指定路由键为 "orange"
    18.         channel.queueBind(queueName, EXCHANGE_NAME, "orange");
    19.  
    20.         System.out.println("等待接收 orange 消息...");
    21.  
    22.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    23.             String message = new String(delivery.getBody(), "UTF-8");
    24.             System.out.println("消费者1接收消息: '" + message + "'");
    25.         };
    26.  
    27.         channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
    28.         });
    29.     }
    30. }
    31. public class LogConsumer2 {
    32.     private final static String EXCHANGE_NAME = "direct_logs";
    33.  
    34.     public static void main(String[] argv) throws Exception {
    35.         ConnectionFactory factory = new ConnectionFactory();
    36.         factory.setHost("123.249.112.12");
    37.         factory.setPort(5672);
    38.  
    39.         Connection connection = factory.newConnection();
    40.         Channel channel = connection.createChannel();
    41.  
    42.         channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
    43.         
    44.         // 创建一个临时队列
    45.         String queueName = channel.queueDeclare().getQueue();
    46.         
    47.         // 绑定队列到交换器，指定路由键为 "black" 和 "green"
    48.         channel.queueBind(queueName, EXCHANGE_NAME, "black");
    49.         channel.queueBind(queueName, EXCHANGE_NAME, "green");
    50.  
    51.         System.out.println("等待接收 black 和 green 消息...");
    52.  
    53.         DeliverCallback deliverCallback = (consumerTag, delivery) -> {
    54.             String message = new String(delivery.getBody(), "UTF-8");
    55.             System.out.println("消费者2接收消息: '" + message + "'");
    56.         };
    57.  
    58.         channel.basicConsume(queueName, true, deliverCallback, consumerTag -> {
    59.         });
    60.     }
    61. }

  • 相关阅读:
    用Python实现基于人脸识别的门禁管理系统(附源码)
    Docker 环境安装postgres+postgis扩展
    什么是构造函数?（JavaScript）
    mysql 不同版本 下载地址
    java 加密结果不一致的各个坑
    Hive mapjoin使用
    光模块厂家如何提高千兆光模块和万兆光模块的可靠性
    使用软路由(openWrt)安装openVPN搭建局域网连接
    基于FPGA的图像直方图统计实现,包括tb测试文件和MATLAB辅助验证
    goroutine 调度器
  • 原文地址：https://blog.csdn.net/weixin_49561506/article/details/133940098