转载RabbitMQ入门(4)--路由

路由

(使用Java客户端)

在先前的指南中,我们建立了一个简单的日志系统。我们可以将我们的日志信息广播到多个接收者。 在这部分的指南中,我们将要往其中添加一个功能-让仅仅订阅一个消息的子集成为可能。例如,我们可以直接将关键的错误信息指向到日志文件(保存在爱硬盘空间),同时依旧能打印所有日志信息到平台上。

绑定

在之前的例子里我们已经创建绑定。你可以回顾下代码:

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

A binding is a relationship between an exchange and a queue. This can be simply read as: the queue is interested in messages from this exchange. 一个绑定是一个交换所和一个队列之间的关系。这个很容易理解为:这个队列是对这 交换机的消息感兴趣。

绑定可以带上额外的路由关键字参数。为了消除对basic_publish参数的迷惑,我们将会将它称之为绑定关键字。以下是我们如何通过一个关键字创建一个绑定:

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

这绑定关键字的意义取决于 交换机类型。这fanout 交换机,我们之前使用的那个,仅仅忽略它的值。

直接交换

我们当前的日志系统将所有消息广播到所有消费者。我们想扩展它,让其允许依据其严格的规则过滤消息。例如我们可能想让一个往硬盘中写日志消息的程序仅仅接收关键的错误,而不是将硬盘空间浪费在警告和信息的日志消息上。 我们使用fanout类型的交换机,那个不会给我们太多的灵活性-它仅仅能胜任没头脑的广播。

我们可以使用direct类型的交换机来替代。一个direct交换机背后的路由算法是简单的-一个消息将会进入那些队列的绑定关键字与消息中路由关键字匹配的队列中。

为了说明那个,考虑接下来结构: 在这个结构里,我们看见了这direct类型的交换机绑定了两个队列。第一个队列装有orange绑定关键字,这第二个有两个绑定,一个是black绑定关键字并且另一个是green关键之。 在这个结构里,发送到交换机里的消息,其中消息中带路由关键字为orange将要路由到队列Q1上,消息中带路由关键字为blackgreen将路由到队列Q2上。所有其他类型的消息会被丢弃。

多种绑定

将一个绑定关键字绑定到货个队列上是十分合法的。在我们例子中使用绑定关键字blackXQ1绑定在一起。既然那样,这direct类型的交换机与fanout类型相似,同样会广播这消息到所有符合的队列中。一个路由关键位balck的关键字将会被传递到Q1Q2

发出日志

我们将会为我们的日志系统使用这个模型。使用direct类型的交换机来代替fanout类型,发送消息。由于这路由关键字我们可以严格的记录。接收程序通过这种方式可以严格接收它想接收的。让我们首先关注发布日志。 总之,我们首先需要创建个交换机。

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

我们准备发送一个消息:

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

为了简化这个事情,我们保证这severityinof,warning,error中的一个。

订阅

接收消息如先前那样工作,有一个例外,我们会把每一个我们感兴趣的severity创建一个新的绑定。

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

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

把它们放在一起

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();

        channel.exchangeDeclare(EXCHANGE_NAME, "direct");

        String severity = getSeverity(argv);
        String message = getMessage(argv);

        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();

        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 + "'");
        }
    }
}

如平常那样编译(看指南第一部分,编译和类路径的建议)。为了方便,当我们运行实例是,我们现在使用一个环境变量$CP(在windows环境上是%CP%)表示类路径。 如果你想仅保存warningerror记录不包含info记录信息到文件里,打开一个控制平台并输入:

$ java -cp $CP ReceiveLogsDirect warning error > logs_from_rabbit.log

如果你想在你的屏幕上看所有的日志信息,打开一个新的终端并键入:

$ java -cp $CP ReceiveLogsDirect info warning error
 [*] Waiting for logs. To exit press CTRL+C

例如,为了发布一个错误日志信息,仅需要键入:

$ java -cp $CP EmitLogDirect error "Run. Run. Or it will explode."
 [x] Sent 'error':'Run. Run. Or it will explode.'

EmitLogDirect.java sourceReceiveLogsDirect.java source的所有源代码。

阅览指南第五部分,查看如何根据一个模式来监听消息。

时间: 2024-07-28 20:53:55

转载RabbitMQ入门(4)--路由的相关文章

转载RabbitMQ入门(3)--发布和订阅

发布和订阅 (使用java 客户端) 在先前的指南中,我们创建了一个工作队列.这工作队列后面的假想是每一个任务都被准确的传递给工作者.在这部分我们将会做一些完全不同的事情–我们将一个消息传递给多个消费者.这部分被认知为"发布和订阅". 为了说明这个部分,我们会建立一个简单德日志系统,它是由两个程序组成–第一个发出日志消息,第二个接收和打印它们. 在我们的日志系统中,每一个运行的接收者拷贝程序将会获得信息.通过这个方式我们可以运行一个接收者,直接的把日志记录到硬盘中:在同一时间我们可以运

转载RabbitMQ入门(2)--工作队列

工作队列 (使用Java客户端) 在这第一指南部分,我们写了通过同一命名的队列发送和接受消息.在这一部分,我们将会创建一个工作队列,在多个工作者之间使用分布式时间任务. 工作队列(亦称:任务队列)背后主要的思想是避免立即处理一个资源密集型任务并且不得不一直等待完成.相反我们可以计划着让任务后续执行.我们将任务封装 成消息,发送到队列中.一个工作者进程在后台运行,获取任务并最终执行任务.当你运行多个工作者,所有的任务将会被他们所共享. 在web应用程序中,这个理念是特别有用的,你无法在一个短暂的h

转载RabbitMQ入门(6)--远程调用

远程过程调用(RPC) (使用Java客户端) 在指南的第二部分,我们学习了如何使用工作队列将耗时的任务分布到多个工作者中. 但是假如我们需要调用远端计算机的函数,等待结果呢?好吧,这又是另一个故事了.这模式通常被称为远程过程调用或RPC. 在这部分,我们将会使用RabbitMQ构建一个RPC系统:一个客户端和一个可扩展的RPC服务器.由于我们还没有值得分散的耗时任务,我们将会创建一个虚拟的RPC服务,用来返回Fibonacci(斐波纳契数列). 用户接口 为了说明RPC服务如何使用,我们将会创

RabbitMQ消息队列(1):RabbitMQ入门

oldriver老司机技术手册 分享 RabbitMQ消息队列(1):RabbitMQ入门

【转载】消息队列RabbitMQ入门介绍

(一)基本概念       RabbitMQ 是流行的开源消息队列系统,用 erlang 语言开发.我曾经对这门语言挺有兴趣,学过一段时间,后来没坚持.RabbitMQ 是 AMQP(高级消息队列协议)的标准实现.如果不熟悉 AMQP,直接看 RabbitMQ 的文档会比较困难.不过它也只有几个关键概念,这里简单介绍. RabbitMQ 的结构图如下: 几个概念说明: Broker:简单来说就是消息队列服务器实体.Exchange:消息交换机,它指定消息按什么规则,路由到哪个队列.Queue:消

07Vue.js快速入门-Vue路由详解

对于前端来说,其实浏览器配合超级连接就很好的实现了路由功能.但是对于单页面应用来说,浏览器和超级连接的跳转方式已经不能适用, 所以各大框架纷纷给出了单页面应用的解决路由跳转的方案. Vue框架的兼容性非常好,可以很好的跟其他第三方的路由框架进行结合.当然官方也给出了路由的方案: vue-router; 建议还是用官方的最好,使用量也是最大,相对来说Vue框架的升级路由组件升级也会及时跟上,所以为了以后的维护和升级方便还是使用Vue自家的东西最好. 7.1. Vue-router的版本对应 注意:

路由器设置教程:PHP框架queryphp教程:入门八 路由跳转

路由说明path_info方式testframework.php?router=default&action=indextestframework.php/default/index可以使用apache重写 去掉testframework.php文件可以自己设置路由规则C("router")->ruleMaps("login",'/login/:id', array('controller' => 'auth', 'action' => '

ThinkPHP 3.1快速入门:路由

ThinkPHP框架对URL有一定的规范,所以如果你希望定制你的URL格式的话,就需要好好了解下内置的路由功能了,它能让你的URL变得更简洁和有文化. 启用路由 要使用路由功能,前提是你的URL支持PATH_INFO,并且在项目配置文件中开启路由: 'URL_ROUTER_ON'&http://www.aliyun.com/zixun/aggregation/37954.html">nbsp;  => true, //开启路由 然后就是配置路由规则了,使用URL_ROUTE_

【转载】关于RabbitMQ的消息持久性

      在消息队列 RabbitMQ 入门介绍里,描述了 RabbitMQ 的持久性设置.在设置持久化后,消息保存在磁盘上,即使 RabbitMQ 重启或服务器重启,消息都不会丢失. RabbitMQ 支持消息的持久化,也就是数据写在磁盘上,为了数据安全考虑,我想大多数用户都会选择持久化.消息队列持久化包括3个部分:(1)exchange 持久化,在声明时指定 durable => 1(2)queue 持久化,在声明时指定 durable => 1(3)消息持久化,在投递时指定 deliv