手机版
你好,游客 登录 注册
背景:
阅读新闻

RabbitMQ学习总结

第五篇:路由Routing

[日期:2016-07-15] 来源:Linux社区  作者:leocook [字体: ]

目录

RabbitMQ学习总结 第一篇:理论篇
RabbitMQ学习总结 第二篇:快速入门HelloWorld

RabbitMQ学习总结 第三篇:工作队列Work Queue

RabbitMQ学习总结 第四篇:发布/订阅 Publish/Subscribe

RabbitMQ学习总结 第五篇:路由Routing

RabbitMQ学习总结 第六篇:Topic类型的exchange

RabbitMQ学习总结 第七篇:RCP(远程过程调用协议)

上一篇中我们构建了一个简单的日志系统,我们可以把日志消息广播给多个接受者。

这篇中我们将来添加一个特性只接收部分消息。例如我只将一些错误log存到文件中,把所有的log都打印到控制台里。

1、绑定(Bindings)

在上篇博文中,我们已经创建了一个binding,代码如下:

channel.queueBind(queueName, EXCHANGE_NAME, "");

一个binding就是exchange和Queue之间的一个关系。可以简单的理解为:这个Queue对其相对于的exchange的消息感兴趣(原文是the queue is interested in messages from this exchange,能理解什么意思,但总觉得怪怪的)。

Binding可以使用一个已经存在的routingKey参数。为了避免和basic_publish参数混淆,我们称之为binding key。下边就是我们怎么用key来创建一个binding:

channel.queueBind(queueName, EXCHANGE_NAME, "black");

binding key的意义有时候取决于exchange的类型。对于Fanout类型的exchange,会忽略binding key。

2、Direct类型的exchange

我们上篇博文中的日志系统会把所有的log消息广播给所有的消费者。我们想扩展来根据他们的日志级别来过滤log消息。例如:我们只想把error级别的日志写到磁盘文件中,而其它级别的日志消息则过滤掉。

我们之前使用的fanout类型的exchange,但这样就不会有太多的灵活性。

在这里我们将要使用direct类型的exchange。Direct类型exchange的路由算法是很简单的:要想一个消息能到达这个队列,需要binding key和routing key正好能匹配得上。

为了说明这个道理,可以看看下边的描述:

在这样的结构中,我们可以看到direct类型的exchange X,有两个queue绑定到它。第一个queue是以orange为binding key绑定到exchange X上的,第二个queue是由两个binding key(black和green)绑定到exchange X的。

在这样的设置中,一条消息被推送到exchange,如果使用的routing key是orange,那么消息就会被路由到Q1中;如果使用的routing key是black或green,那么该消息将会被路由到Q2中。其它的消息都将会被丢弃掉。

3、多重绑定(Multiple bindings)

用同一个binding来把多个queue绑定到同一个exchange也是可行的。例如在之前例子的基础上,在X和Q1之间添加binding key名字为black,这样的话,这里的direct类型的exchange就和fanout类型的一样了,可以把消息推送给所有的queue。带有routing key为black的消息将会被推送到Q1和Q2中。

4、发送日志(Emitting logs)

我们将会使用这种模型,不使用fanout类型的exchange,而是使用direct类型的。我们使用日志级别做为routing key,接收端根据设置的日志级别做为binding key来接收消息。首先来看看发射日志:

如之前一样,首先来创建一个exchange:

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

然后准备发送消息;

channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());

这里的”severity”可以是”info”、“warning”、”error”等。

5、订阅(Subscribing)

这里接收消息和上篇博文中的一样,只是有一点例外:我们将会为每一个感兴趣的日志级别进行绑定。

String queueName = channel.queueDeclare().getQueue();

for(String severity : argv){   
  channel.queueBind(queueName, EXCHANGE_NAME, severity);
}

6、最终实现

  • EmitLogDirect.Java的代码:
public class EmitLogDirect {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] argv)
                  throws java.io.IOException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
 
        //声明direct类型的exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
                  
    
//拿到日志级别 String severity = getSeverity(argv); //拿到日志消息 String message = getMessage(argv); //指定routing key,发送消息 channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes()); System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
channel.close(); connection.close(); }
//.. }
  • ReceiveLogsDirect.java的代码:
public class ReceiveLogsDirect {
    private static final String EXCHANGE_NAME = "direct_logs";
    public static void main(String[] argv)
                  throws java.io.IOException,
                  java.lang.InterruptedException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        //声明direct类型的exchange
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        String queueName = channel.queueDeclare().getQueue();

        if (argv.length < 1){
            System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
            System.exit(1);
        }
        
        //绑定我们需要接收的日志级别
        for(String severity : argv){
            channel.queueBind(queueName, EXCHANGE_NAME, severity);
        }

        System.out.println(" [*] Waiting for 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 + "':'" + message + "'");
        }
    }
}
  • 运行三个日志接收器:

接收error和info级别的日志:

接收error级别的日志:

接收info级别的日志:

  • 运行两个日志发生器:

产生error级别的日志:

产生info级别的日志:

  • 观察接收器端的变化:

接收error级别的接收器,只接收error级别的日志:

 

接收info级别的接收器,只接收info级别的日志:

 

Error和info级别日志都接收的接收器,info和error级别的日志都接收:

7、总结:

要记住生产者端的routing key,那么在消费者端设置binding key和之前的routing key一样,就可以用direct类型的exchange了,以此来获取到自己需要的消息。

参考链接:http://www.rabbitmq.com/tutorials/tutorial-four-java.html

linux
【内容导航】
第1页:第一篇:理论篇 第2页:第二篇:快速入门HelloWorld
第3页:第三篇:工作队列Work Queue 第4页:第四篇:发布/订阅 Publish/Subscribe
第5页:第五篇:路由Routing 第6页:第六篇:Topic类型的exchange
第7页:第七篇:RCP(远程过程调用协议)
相关资讯       RabbitMQ 
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数

       

评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款