（转）RabbitMQ学习之主题topic（java）

http://blog.csdn.net/zhu_tianwei/article/details/40887775

参考：http://blog.csdn.NET/lmj623565791/article/details/37706355

direct类型的消息通过绑定键转发到队列，但是存在一些局限性：它不能够基于多重条件进行路由选择，有可能希望不仅根据日志的级别而且想根据日志的来源进行订阅，这就需要主题类型的转发器来实现。

发往主题类型的转发器的消息不能随意的设置选择键（routing_key），必须是由点隔开的一系列的标识符组成。标识符可以是任何东西，但是一般都与消息的某些特性相关。一些合法的选择键的例子："stock.usd.nyse", "nyse.vmw","quick.orange.rabbit".你可以定义任何数量的标识符，上限为255个字节。
绑定键和选择键的形式一样。主题类型的转发器背后的逻辑和直接类型的转发器很类似：一个附带特殊的选择键将会被转发到绑定键与之匹配的队列中。需要注意的是：关于绑定键有两种特殊的情况。
*可以匹配一个标识符。
#可以匹配0个或多个标识符。

1.发送日志消息SendLogTopic，发送4个消息绑定不同的绑定键， "kernal.info", "cron.warning", "auth.info", "kernel.critical"

[java] view plaincopy print?

package cn.slimsmart.rabbitmq.demo.topic;
import java.util.UUID;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
//发送消息端
public class SendLogTopic {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws Exception {
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.101.174");
factory.setUsername("admin");
factory.setPassword("admin");
factory.setPort(AMQP.PROTOCOL.PORT);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明转发器
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
//定义绑定键
String[] routing_keys = new String[] { "kernal.info", "cron.warning",
"auth.info", "kernel.critical" };
for (String routing_key : routing_keys)
{
//发送4条不同绑定键的消息
String msg = UUID.randomUUID().toString();
channel.basicPublish(EXCHANGE_NAME, routing_key, null, msg
.getBytes());
System.out.println(" [x] Sent routingKey = "+routing_key+" ,msg = " + msg + ".");
}
channel.close();
connection.close();
}
}

2.定义接收kernel.*消息的消费者

[java] view plaincopy print?

package cn.slimsmart.rabbitmq.demo.topic;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
//接收kernel.*消息
public class ReceiveLogsTopicForKernel {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws Exception {
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.101.174");
factory.setUsername("admin");
factory.setPassword("admin");
factory.setPort(AMQP.PROTOCOL.PORT);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明转发器
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// 随机生成一个队列
String queueName = channel.queueDeclare().getQueue();
//接收所有与kernel相关的消息
channel.queueBind(queueName, EXCHANGE_NAME, "kernel.*");
System.out.println(" [*] Waiting for messages about kernel. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true)
{
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(" [x] Received routingKey = " + routingKey
+ ",msg = " + message + ".");
}
}
}

3.接收*.critical消息消费者

[java] view plaincopy print?

package cn.slimsmart.rabbitmq.demo.topic;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.QueueingConsumer;
//接收*.critical消息
public class ReceiveLogsTopicForCritical {
private static final String EXCHANGE_NAME = "topic_logs";
public static void main(String[] args) throws Exception {
// 创建连接和频道
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("192.168.101.174");
factory.setUsername("admin");
factory.setPassword("admin");
factory.setPort(AMQP.PROTOCOL.PORT);
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();
// 声明转发器
channel.exchangeDeclare(EXCHANGE_NAME, "topic");
// 随机生成一个队列
String queueName = channel.queueDeclare().getQueue();
// 接收所有与kernel相关的消息
channel.queueBind(queueName, EXCHANGE_NAME, "*.critical");
System.out
.println(" [*] Waiting for critical messages. To exit press CTRL+C");
QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(queueName, true, consumer);
while (true)
{
QueueingConsumer.Delivery delivery = consumer.nextDelivery();
String message = new String(delivery.getBody());
String routingKey = delivery.getEnvelope().getRoutingKey();
System.out.println(" [x] Received routingKey = " + routingKey
+ ",msg = " + message + ".");
}
}
}

启动2个消费者，再启动发送4类消息生产者。观察接收到的消息，都收到对应的消息。可以看出使用topic类型的转发器，成功实现了多重条件选择的订阅。

转载于:https://www.cnblogs.com/telwanggs/p/7124621.html

（转）RabbitMQ学习之主题topic（java）相关推荐

（转） RabbitMQ学习之helloword（java）
http://blog.csdn.net/zhu_tianwei/article/details/40835555 amqp-client:http://www.rabbitmq.com/java-c ...
（转） RabbitMQ学习之工作队列（java）
http://blog.csdn.net/zhu_tianwei/article/details/40887717 参考:http://blog.csdn.NET/lmj623565791/artic ...
（转）RabbitMQ学习之路由（java）
http://blog.csdn.net/zhu_tianwei/article/details/40887755 参考:http://blog.csdn.NET/lmj623565791/artic ...
使用java实现MQTT协议客户端的接收、发布消息和订阅、退订主题topic
记录一下我实习的第一个任务,学习MQTT协议首先呢得了解MQTT是什么,这里推荐一个我学习MQTT的中文文档 MQTT协议的基于TCP/IP协议的一个物联网协议,有几个概念必须要弄懂得主题(topi ...
RabbitMQ 学习笔记
RabbitMQ 学习笔记 RabbitMQ 学习笔记 1. 中间件 1.1 什么是中间件 1.2 为什么要使用消息中间件 1.3 中间件特点 1.4 在项目中什么时候使用中间件技术 2. 中间件技术 ...
Rabbitmq学习笔记(尚硅谷2021)
Rabbitmq学习笔记 (尚硅谷) 1.MQ 的概念 1.1 什么是 MQ? 1.2 为什么要用 MQ? 削峰解耦异步 1.3 MQ 的分类 ActiveMQ Kafka RocketMQ Ra ...
RabbitMQ学习笔记
目录一.MQ 的相关概念 MQ是什么? MQ三大优势 MQ的劣势 MQ 的产品 RabbitMQ核心 JMS 安装二.HelloWorld 三.Work Queues(轮训) 消息应答 Rabbi ...
乐行学院RabbitMQ学习教程第一章 RabbitMQ介绍(可供技术选型时使用)
乐行学院RabbitMQ学习教程第一章 RabbitMQ介绍 RabbitMQ介绍 1.RabbitMQ技术简介 2.RabbitMQ其他扩展插件 2.1监控工具rabbitmq-managemen ...
Rabbitmq学习笔记教程-尚硅谷
Rabbitmq学习笔记 (尚硅谷) 尚硅谷 rabbitmq 教程 1.MQ 的概念 1.1 什么是 MQ? 存放消息的队列,互联网架构中常见的一种服务与服务之间通信的方式. 1.2 为什么要用 M ...

（转）RabbitMQ学习之主题topic（java）

（转）RabbitMQ学习之主题topic（java）相关推荐

最新文章

热门文章