RabbitMQ交换机(3)-Topic
作者:mmseoamin日期:2024-01-19

1.Topic模式

RabbitMQ的Topic模式是一种基于主题的消息传递模式。它允许发送者向一个特定的主题(topic)发布消息,同时,订阅者也可以针对自己感兴趣的主题进行订阅。

在Topic模式中, 主题通过一个由单词和点号组成的字符串来描述。例如,“#.china”表示匹配所有以“china”为结尾的主题,比如“bj.china”或“china”等。( ‘ # ’ 和 ‘ * ’ 会再后面介绍)

当一个消息被发布到Topic交换机(Exchange)时,交换机会将消息转发给所有与该主题匹配的队列。消费者(即订阅者)可以对队列进行绑定,通过指定自己感兴趣的主题进行绑定。

通过使用Topic模式,我们可以实现高度灵活的信息交换模式,同时,确保只有感兴趣的消费者才会收到消息,提高了系统的效率和可靠性。

  1. *匹配1个
  2. #匹配0个或多个
  3. 当一个队列绑定键是#,那么这个队列将接收所有数据,就有点像 fanout 了
  4. 如果队列绑定键当中没有#和*出现,那么该队列绑定类型就是 direct 了

    RabbitMQ交换机(3)-Topic,在这里插入图片描述,第1张

2. 生产者

RabbitMQ交换机(3)-Topic,在这里插入图片描述,第2张

RabbitMQ交换机(3)-Topic,在这里插入图片描述,第3张

package com.hong.rabbitmq8;
import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
/**
 * @Description: Topic模式生产者
 * @Author: hong
 * @Date: 2024-01-16 20:09
 * @Version: 1.0
 **/
public class TopicSend {
    private static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        Map bindingKeyMap = new HashMap<>();
        bindingKeyMap.put("quick.orange.rabbit", "被队列 Q1Q2 接收到");
        bindingKeyMap.put("lazy.orange.elephant", "被队列 Q1Q2 接收到");
        bindingKeyMap.put("quick.orange.fox", "被队列 Q1 接收到");
        bindingKeyMap.put("lazy.brown.fox", "被队列 Q2 接收到");
        bindingKeyMap.put("lazy.pink.rabbit", "虽然满足两个绑定但只被队列 Q2 接收一次");
        bindingKeyMap.put("quick.brown.fox", "不匹配任何绑定不会被任何队列接收到会被丢弃");
        bindingKeyMap.put("quick.orange.male.rabbit", "是四个单词不匹配任何绑定会被丢弃");
        bindingKeyMap.put("lazy.orange.male.rabbit", "是四个单词但匹配 Q2");
        for (Map.Entry bindingKeyEntry : bindingKeyMap.entrySet()) {
            String bindingKey = bindingKeyEntry.getKey();
            String message = bindingKeyEntry.getValue();
            channel.basicPublish(EXCHANGE_NAME, bindingKey, null, message.getBytes(StandardCharsets.UTF_8));
            System.out.println("消息发送完成------" +bindingKey+ message);
        }
    }
}

3.消费者1

package com.hong.rabbitmq8;
import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
/**
 * @Description: Topic模式接受者1-接收*.orange.*
 * @Author: hong
 * @Date: 2024-01-16 20:07
 * @Version: 1.0
 **/
public class TopicReceiver1 {
    public static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        String queueName = "Q1";
        channel.queueDeclare(queueName,false,false,false,null);
        channel.queueBind(queueName,EXCHANGE_NAME,"*.orange.*");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
            System.out.println("接收队列:" + queueName + ",routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };
        CancelCallback cancelCallback = var -> {
        };
        channel.basicConsume(queueName,true,deliverCallback,cancelCallback);
    }
}

4.消费者2

package com.hong.rabbitmq8;
import com.hong.utils.RabbitMQUtil;
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.CancelCallback;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DeliverCallback;
import java.nio.charset.StandardCharsets;
/**
 * @Description: Topic模式接受者1-接收*.*.rabbit和lazy.#
 * @Author: hong
 * @Date: 2024-01-16 20:07
 * @Version: 1.0
 **/
public class TopicReceiver2 {
    public static final String EXCHANGE_NAME = "topic_logs";
    public static void main(String[] args) throws Exception {
        Channel channel = RabbitMQUtil.getChannel();
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.TOPIC);
        String queueName = "Q2";
        channel.queueDeclare(queueName,false,false,false,null);
        channel.queueBind(queueName,EXCHANGE_NAME,"*.*.rabbit");
        channel.queueBind(queueName,EXCHANGE_NAME,"lazy.#");
        DeliverCallback deliverCallback = (comsumerTag, message) -> {
            System.out.println("接收队列:" + queueName + ",routingKey:" + message.getEnvelope().getRoutingKey() + ",消息:" + new String(message.getBody(), StandardCharsets.UTF_8));
        };
        CancelCallback cancelCallback = var -> {
        };
        channel.basicConsume(queueName,true,deliverCallback,cancelCallback);
    }
}

RabbitMQ交换机(3)-Topic,在这里插入图片描述,第4张

RabbitMQ交换机(3)-Topic,在这里插入图片描述,第5张

RabbitMQ交换机(3)-Topic,在这里插入图片描述,第6张

例子说明
quick.orange.rabbit被队列 Q1Q2 接收到
lazy.orange.elephant被队列 Q1Q2 接收到
quick.orange.fox被队列 Q1 接收到
lazy.brown.fox被队列 Q2 接收到
lazy.pink.rabbit虽然满足两个绑定但只被队列 Q2 接收一次
quick.brown.fox不匹配任何绑定不会被任何队列接收到会被丢弃
quick.orange.male.rabbit是四个单词不匹配任何绑定会被丢弃
lazy.orange.male.rabbit是四个单词但匹配 Q2