概述

上篇direct必须是生产者发布消息指定的routingKey和消费者在队列绑定时指定的routingKey完全相等时才能匹配到队列上。topic与direct不同,可以进行模糊匹配。topic交换机的消息不能随意的设置选择routingKey，必须是由点隔开的一系列的标识符组成。标识符可以是任何东西，但是一般都与消息的某些特性相关。例子：“link.orange.rabbit”,你可以定义任何数量的标识符，上限为255个字节

匹配规则：

* 其中星号可以代替一个单词

# 可以替代零个或更多的单词，只要能模糊匹配上就能将消息映射到队列中。当一个队列的绑定键为#的时候，这个队列将会无视消息的路由键，接收所有的消息

注：队列1 将接受warn相关的消息

队列2 将接受info 相关消息 debugger所有消息

代码

生产者

package com.ll.mq.hellomq.topic;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;/*** * @author ll 生产者**/
public class Producer {public static void main(String[] args) {try {ConnectionFactory factory = new ConnectionFactory();factory.setHost("127.0.0.1");factory.setPort(5672);factory.setUsername("kysc");factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// Routing 的路由规则使用直连接String EXCHANGE_NAME = "exchange.topic";String[] routingKeys = { "link.warn.rabbit", "link.warn.elephant", "debugger.info"};for (String routingKey : routingKeys) {String message = "Hello RabbitMQ - " + routingKey;System.out.println(message);channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));}// 关闭资源channel.close();connection.close();} catch (Exception e) {e.printStackTrace();}}
}

消费者1

package com.ll.mq.hellomq.topic;import java.io.IOException;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/*** * @author ll 消费者1**/
public class ConsumerOne {static String EXCHANGE_NAME = "exchange.topic";static String QUEUE_NAME = "queue.topic.q1";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);factory.setUsername("kysc");factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明交换机类型channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.exchangeDeclare(EXCHANGE_NAME, "topic");String[] routingKeys = {"*.warn.*"};for (int i = 0; i < routingKeys.length; i++) {channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, routingKeys[i]);}Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println(" [C] ConsumerOne '" + message + "', 接受...");}};channel.basicConsume(QUEUE_NAME, true, consumer);}
}

消费者2

package com.ll.mq.hellomq.topic;import java.io.IOException;import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
/*** * @author ll 消费者2**/
public class ConsumerTwo {static String EXCHANGE_NAME = "exchange.topic";static String QUEUE_NAME = "queue.topic.q2";public static void main(String[] argv) throws Exception {ConnectionFactory factory = new ConnectionFactory();factory.setHost("localhost");factory.setPort(5672);factory.setUsername("kysc");factory.setPassword("123456");Connection connection = factory.newConnection();Channel channel = connection.createChannel();// 声明交换机类型channel.queueDeclare(QUEUE_NAME, false, false, false, null);channel.exchangeDeclare(EXCHANGE_NAME, "topic");String[] routingKeys = {"debugger.#"};for (int i = 0; i < routingKeys.length; i++) {channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, routingKeys[i]);}Consumer consumer = new DefaultConsumer(channel){@Overridepublic void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {String message = new String(body, "UTF-8");System.out.println(" [C] ConsumerTwo '" + message + "', 接受中...");}};channel.basicConsume(QUEUE_NAME, true, consumer);}
}

结果

生产者：Hello RabbitMQ - link.warn.rabbit
Hello RabbitMQ - link.warn.elephant
Hello RabbitMQ - debugger.info

消费者1 [C] ConsumerOne 'Hello RabbitMQ - link.warn.rabbit', 接受...
[C] ConsumerOne 'Hello RabbitMQ - link.warn.elephant', 接受...

消费者2 [C] ConsumerTwo 'Hello RabbitMQ - debugger.info', 接受中...

下一篇 https://blog.csdn.net/lilongwangyamin/article/details/105118955 rabbitmq rpc远程过程调用

RabbitMq topic相关推荐

rabbitmq+topic+java_译:5.RabbitMQ Java Client 之 Topics （主题）
我们使用的是direct(直接交换),而不是使用只能进行虚拟广播的 fanout(扇出交换),并且有可能选择性地接收日志. 虽然使用direct(直接交换)改进了我们的系统,但它仍然有局限性 - 它不 ...
rabbitmq topic 收不到数据_RabbitMQ和Kafka到底怎么选？
作者:haolujunhttp://cnblogs.com/haolujun/p/9632835.html 前言开源社区有好多优秀的队列中间件,比如RabbitMQ和Kafka,每个队列都貌似有其特 ...
RabbitMQ Topic exchange
Topic exchange topic与之前的每个类型都不同(ps:废话每个都是不同的).Topic解决了我们另一个需求.举个例子,有一个做资讯的公司,他们会收集各种科技公司的动态并且第一时间转发出 ...
RabbitMQ Topic交换机（生产者）
/* * 声明主题交换机及相关队列 * 消费者C1 * */ public class ReceiveLogsTopic01 {//交换机名称public static final String EX ...
RabbitMQ Topic交换机的作用
介绍发送到类型是topic交换机的消息的routing_key不能随意写,必须满足一定的要求,它必须是一个单词列表,以点号分隔开.这些单词可以是任意单词,比如说: "stock.usd.n ...
RabbitMQ Topic模式
一个生产者,一个交换机,两个队列,两个消费者生产者创建Topic的exchange并且绑定到队列中,这次绑定可以通过*和#关键字,对指定RoutingKey内容,编写时注意格式 xxx.xxx.xx ...
RabbitMQ Topic交换机（结果成功）
public class EmitLogTopic {//交换机的名称public static final String EXCHANGE_NAME = "topic_logs" ...
RabbitMq(七) Topic模式介绍及代码示例
概述: 在上一文章中我们介绍了路由模式(Routing),routing模式是不同的消息队列绑定了不同的路由key,但是我们看出路由key为固定的字符串标记.而本章中的Topic模式则为在路由模式下, ...
Springboot整合RabbitMQ,包含direct,topic,fanout三种模式的整合
一 RabbitMQ的介绍 RabbitMQ是消息中间件的一种,消息中间件即分布式系统中完成消息的发送和接收的基础软件.这些软件有很多,包括ActiveMQ(apache公司的),RocketMQ(阿 ...

RabbitMq topic

概述

代码

结果

RabbitMq topic相关推荐

最新文章

热门文章