RabbitMQ学习之主题topic(java) - Go语言中文社区

RabbitMQ学习之主题topic(java)


参考: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" 

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.*消息的消费者

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消息消费者

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类型的转发器,成功实现了多重条件选择的订阅。


版权声明:本文来源CSDN,感谢博主原创文章,遵循 CC 4.0 by-sa 版权协议,转载请附上原文出处链接和本声明。
原文链接:https://blog.csdn.net/tianwei7518/article/details/40887775
站方申明:本站部分内容来自社区用户分享,若涉及侵权,请联系站方删除。
  • 发表于 2020-03-08 05:38:54
  • 阅读 ( 853 )
  • 分类:

0 条评论

请先 登录 后评论

官方社群

GO教程

猜你喜欢