Java使用RabbitMQ之订阅分发（Topic）

Myth丶恋晨 2022-10-22 09:09 26阅读 0赞

使用RabbitMQ进行消息发布和订阅，生产者将消息发送给转发器（exchange），转发器根据路由键匹配已绑定的消息队列并转发消息，主题模式支持路由键的通配。

*  生产者代码：

1 package org.study.exchange3.topic3;
     2 
     3 import com.rabbitmq.client.Channel;
     4 import com.rabbitmq.client.Connection;
     5 import org.junit.Test;
     6 import org.study.utils.ConnectionUtils;
     7 
     8 import java.io.IOException;
     9 import java.util.concurrent.TimeoutException;
    10 
    11 /**
    12  * topic-主题模式（分发订阅）
    13  * exchange只转发消息，但是没有存储能力，只有队列才有存储能力
    14  * 主题模式支持路由键的通配符
    15  * “#”表示0个或若干个关键字，“*”表示一个关键字。
    16  */
    17 public class Sender {
    18     public static final String QUEUE_NAME = "test_topic_queue";
    19     public static final String EXCHANGE_NAME = "topic_exchange";
    20 
    21     @Test
    22     public void send() throws IOException, TimeoutException, InterruptedException {
    23         // 获取连接
    24         Connection conn = ConnectionUtils.getConnection();
    25         // 获取通道
    26         Channel channel = conn.createChannel();
    27 //        //创建队列
    28 //        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    29         //声明转发器
    30         channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    31         //每个消费者发送确认消息前，只发送一条消息
    32         channel.basicQos(1);
    33 
    34         String msg = "hello rabbitmq topic !";
    35         //发送消息至转发器,指定路由键
    36         channel.basicPublish(EXCHANGE_NAME, "key.key", null, msg.getBytes());
    37         System.out.println("[send] msg " + msg);
    38 
    39         channel.close();
    40         conn.close();
    41     }
    42 }

*  消费者代码：

1 package org.study.exchange3.topic3;
     2 
     3 import com.rabbitmq.client.*;
     4 import org.junit.Test;
     5 import org.study.utils.ConnectionUtils;
     6 
     7 import java.io.IOException;
     8 import java.util.concurrent.TimeoutException;
     9 
    10 /**
    11  * 主题模式-接收消息
    12  */
    13 public class Recv {
    14     public static final String QUEUE_NAME = "test_topic_queue";
    15     public static final String EXCHANGE_NAME = "topic_exchange";
    16 
    17     @Test
    18     public void recv() throws IOException, TimeoutException, InterruptedException {
    19         Connection conn = ConnectionUtils.getConnection();
    20         Channel channel = conn.createChannel();
    21         channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    22         channel.basicQos(1);
    23         /*
    24          * 队列绑定转发器,路由键通配符#和*
    25          * #：表示0个或多个字符
    26          * *：表示一个字符
    27          * */
    28         channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, "key.#");
    29 
    30         //定义消费者
    31         DefaultConsumer consumer = new DefaultConsumer(channel) {
    32             //重写获取到达消息
    33             @Override
    34             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    35                 String msg = new String(body, "utf-8");
    36                 System.out.println("[1] recv: " + msg);
    37 
    38                 try {
    39                     Thread.sleep(100);
    40                 } catch (InterruptedException e) {
    41                     e.printStackTrace();
    42                 } finally {
    43                     System.out.println("[1] done!");
    44                     // 回执
    45                     channel.basicAck(envelope.getDeliveryTag(), false);
    46                 }
    47             }
    48         };
    49 
    50         while (true) {
    51             //监听队列
    52             channel.basicConsume(QUEUE_NAME, false, consumer);
    53             Thread.sleep(1000);
    54         }
    55 
    56     }
    57 }