RabbitMQ(四):Exchange交换器–direct

RabbitMQ(四):Exchange交换器–direct

内容翻译自:RabbitMQ Tutorials Java版


RabbitMQ(一):Hello World程序

RabbitMQ(二):Work Queues、循环分发、消息确认、持久化、公平分发

RabbitMQ(三):Exchange交换器–fanout

RabbitMQ(四):Exchange交换器–direct

RabbitMQ(五):Exchange交换器–topic

RabbitMQ(六):回调队列callback queue、关联标识correlation id、实现简单的RPC系统

RabbitMQ(七):常用方法说明 与 学习小结


Routing:

在上一篇博客中,我们创建了一个简单的日志系统。我们可以将日志消息广播给所有的接收者(消费者)。

在这个教程中,我们将为我们的日志系统添加一个功能:仅仅订阅一部分消息。比如,我们可以直接将关键的错误类型日志消息保存到日志文件中,还可以同时将所有的日志消息打印到控制台。


绑定(Bindings):

在之前的例子中,我们已经创建了绑定:

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

一个绑定是建立在一个队列和一个路由器之间的关系,可以解读为:该队列对这个路由器中的消息感兴趣。

绑定可以设置另外的参数:路由键routingKey。为了避免和void basicPublish(String exchange, String routingKey, BasicProperties props, byte[] body)中的routingKey混淆,我们将这里的key称为绑定键binding key,下面的代码展示了如何使用绑定键来创建一个绑定关系:

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

绑定键的含义取决于路由器的类型,我们之前使用的fanout类型路由器会忽略该值。


直接路由器 (Direct Exchange):

我们之前的日志系统会将所有消息广播给所有消费者。现在我们想根据日志的严重程度来过滤日志。比如,我们想要一个程序来将error日志写到磁盘文件中,而不要将warninginfo日志写到磁盘中,以免浪费磁盘空间。

我们之前使用的fanout路由器缺少灵活性,它只是没头脑地广播消息。所以,我们用direct路由器来替换它。direct路由器背后的路由算法很简单:只有当消息的路由键routing key与队列的绑定键binding key完全匹配时,该消息才会进入该队列。

为了演示上面拗口的表述中的意思,考虑下面的设置:

RabbitMQ(四):Exchange交换器--direct

上图中,直接路由器X与两个队列绑定。第一个队列以绑定键orange来绑定,第二个队列以两个绑定键blackgreen和路由器绑定。

按照这种设置,路由键为orange的消息在发布给路由器后,将会被路由到队列Q1,路由键为black或者green的消息将会路由到队列Q2。


多重绑定(Multiple bindings):

RabbitMQ(四):Exchange交换器--direct

多个队列以相同的绑定键binding key绑定到同一个Exchange上,是完全可以的。按照这种方式设置的话,直接路由器就会像fanout路由器一样,将消息广播给所有符合路由规则的队列。一个路由键为black的消息将会发布到队列Q1和Q2。


发布消息:

在这个教程中,我们使用direct路由器来代替上个教程中的fanout路由器。同时,我们为日志设置严重级别,并将此作为路由键。这样,接收者(消费者)就可以选择性地接收日志消息。

首先,创建一个路由器:

channel.exchangeDeclare(EXCHANGE_NAME, "direct");

接着,发送一个消息:

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

简单起见,我们假设severity只能是infowarningerror中的一种。


消息订阅:

接收消息将会和之前的教程类似,只是我们会为每一个级别的消息来创建不同的绑定:

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

放在一块:

RabbitMQ(四):Exchange交换器--direct

生产者EmitLogDirect.java的完整代码:

import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;

public class EmitLogDirect {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv) throws Exception {
        //创建连接
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //声明路由器和路由器的类型
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        
        String severity = "info";
        String message = ".........i am msg.........";

        //发布消息
        channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));
        System.out.println(" [x] Sent '" + severity + "':'" + message + "'");

        channel.close();
        connection.close();
    }
    
}

消费者ReceiveLogsDirect.java的完整代码如下:

import com.rabbitmq.client.*;
import java.io.IOException;

public class ReceiveLogsDirect {

    private static final String EXCHANGE_NAME = "direct_logs";

    public static void main(String[] argv) throws Exception {
        //建立连接和通道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //声明路由器和类型
        channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.DIRECT);
        //声明队列
        String queueName = channel.queueDeclare().getQueue();
        //定义要监听的级别
        String[] severities = {"info", "warning", "error"};
        //根据绑定键绑定
        for (String severity : severities) {
            channel.queueBind(queueName, EXCHANGE_NAME, severity);
        }
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        Consumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                                       AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body, "UTF-8");
                System.out.println(" [x] Received '" + envelope.getRoutingKey() + "':'" + message + "'");
            }
        };
        channel.basicConsume(queueName, true, consumer);
    }
}

现在可以进行测试了。首先,启动一个消费者实例(ReceiveLogsDirect.java),然后将其中的要监听的级别改为String[] severities = {"error"};,再启动另一个消费者实例。此时,这两个消费者都开始监听了,一个监听所有级别的日志消息,另一个监听error日志消息。
然后,启动生产者(EmitLogDirect.java),之后将String severity = "info";中的info,分别改为warningerror后运行。
这样,就可以在控制台看到如下输出:

//生产者
[x] Sent 'warning':'.........i am msg.........'
[x] Sent 'info':'.........i am msg.........'
[x] Sent 'error':'.........i am msg.........'
//消费者1
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'info':'.........i am msg.........'
 [x] Received 'error':'.........i am msg.........'
 [x] Received 'warning':'.........i am msg.........'
//消费者2
 [*] Waiting for messages. To exit press CTRL+C
 [x] Received 'error':'.........i am msg.........'

 


说明:

①与原文略有出入,如有疑问,请参阅原文

②原文均是编译后通过javacp命令直接运行程序,我是在IDE中进行的,相应的操作做了修改。

版权声明:本文内容由互联网用户自发贡献,该文观点仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 举报,一经查实,本站将立刻删除。

发布者:全栈程序员-用户IM,转载请注明出处:https://javaforall.cn/114606.html原文链接:https://javaforall.cn

【正版授权,激活自己账号】: Jetbrains全家桶Ide使用,1年售后保障,每天仅需1毛

【官方授权 正版激活】: 官方授权 正版激活 支持Jetbrains家族下所有IDE 使用个人JB账号...

(0)


相关推荐

  • linux文件共享 samba_文件共享服务

    linux文件共享 samba_文件共享服务Samba是在Linux和UNIX系统上实现SMB协议的一个免费软件,由服务器及客户端程序构成;SMB(ServerMessagesBlock,信息服务块)是一种在局域网上共享文件和打印机的一种通信协议,它为局域网内的不同计算机之间提供文件及打印机等资源的共享服务;SMB协议是客户机/服务器型协议,客户机通过该协议可以访问服务器上的共享文件系统,

  • 举例说明一下怎么算是第一范式、第二范式、第三范式?

    举例说明一下怎么算是第一范式、第二范式、第三范式?数据库的设计范式是数据库设计所需要满足的规范,满足这些规范的数据库是简洁的、结构明晰的,同时,不会发生插入(insert)、删除(delete)和更新(update)操作异常。反之则是乱七八糟,不仅给数据库的编程人员制造麻烦,而且面目可憎,可能存储了大量不需要的冗余信息。设计范式是不是很难懂呢?非也,大学教材上给我们一堆数学公式我们当然看不懂,也记不住。所以我们很多人就根本不按照范式来设计数据库。

  • 爆炸人游戏各关的道具_盗版星露谷可以联机吗

    爆炸人游戏各关的道具_盗版星露谷可以联机吗1.算法流程2.update函数运行流程updateDelta:Updatedeltaifweneedittocheckrelinearizationlater update.pushBackFactors:Addanynewfactors 为每一个新的factor产生一个索引,把新factor装入nonlinearFactors中 把需要移除的factor从nonlinearFactors和linearFactors中同时移除,把从nonlinea.

    2022年10月26日
  • IJ快捷键

    IJ快捷键ctrl+shift+alt:多行操作psvm:生成main()方法;fori:生成for循环;Ctrl+Alt+v:自动补齐返回值类型ctrl+o:覆写方法ctrl+i:实现接口中的方法ctrl+shift+u:大小写转换CTRL+SHIFT+Z:取消撤销Alt+Insert:生成构造方法、getter、setterctrl+y:删除当前行Ctrl+Shift+J:将选中的行合并成一行ctrl+g:定位到某一行Ctrl+Shitft+向下箭头:将光标所在的代码块向下整体移动Ct.

  • Nginx搭建视频点播和视频直播服务器

    Nginx搭建视频点播和视频直播服务器Nginx搭建视频点播和视频直播服务器一·、环境:Centos7,(推荐,Ubuntu不是很好用,经常会有一些莫名其妙的报错)Nginx1.10.1二、系统环境搭建首先,我是不建议自己一个个去安装这些软件的,耗时耗力,而且,容易出错,所以,最好使用yuminstall***命令安装,出错的概率小。资源链接:链接:https://pan.baidu.com/s/1WmJYpQ_b…

  • Unity3D开发入门教程(一)——搭建开发环境[通俗易懂]

    Unity3D开发入门教程(一)——搭建开发环境[通俗易懂]五邑隐侠,本名关健昌,12年游戏生涯。本教程以Unity3D+VSCode+C#+xlua为例。一、安装Unity1、从官网下载UnityHub:https://uni

发表回复

您的电子邮箱地址不会被公开。

关注全栈程序员社区公众号