轻松搞定RabbitMQ(四)——发布/订阅
2017-09-18 00:00
357 查看
轻松搞定RabbitMQ(四)——发布/订阅
在前面的教程中,我们创建了一个工作队列,都是假设一个任务只交给一个消费者。这次我们做一些完全不同的事儿——将消息发送给多个消费者。这种模式叫做“发布/订阅”。为了说明这个模式,我们将构建一个简单日志系统。它包含2段程序:第一个将发出日志消息,第二个接受并打印消息。
如果在日志系统中每一个接受者(订阅者)都会的得到消息的拷贝。那样的话,我们可以运行一个接受者(订阅者)程序,直接把日志记录到硬盘。同时运行另一个接受者(订阅者)程序,打印日志到屏幕上。
说白了,发表日志消息将被广播给所有的接收者。
Exchanges(转发器)
前面的博文汇总,我们都是基于一个队列发送和接受消息。现在介绍一下完整的消息传递模式。RabbitMQ消息模式的核心理念是:生产者没有直接发送任何消费到队列。实际上,生产者都不知道这个消费是发送给哪个队列的。
相反,生产者只能发送消息给转发器,转发器是非常简单的。一方面它接受生产者的消息,另一方面向队列推送消息。转发器必须清楚的知道如何处理接收到的消息。附加一个特定的队列吗?附加多个队列?或者是否丢弃?这些规则通过转发器的类型进行定义。
类型有:Direct、Topic、Headers和Fanout。我们关注最后一个。现在让我们创建一个该类型的转发器,定义如下:
channel.exchangeDeclare("logs", "fanout");
fanout:转发器非常简单,从名字就可以看出,它是广播接受到的消息给所有的队列。而这正好符合日志系统的需求。
Nameless exchange(匿名转发) direct 类型的
之前我们对转换器一无所知,却可以将消息发送到队列,那是可能是我们用了默认的转发器,转发器名为空字符串""。之前我们发布消息的代码是:channel.basicPublish("", "hello", null, message.getBytes());
第一个参数就是转发器的名字,空字符串表示模式或者匿名的转发器。消息通过队列的routingKey路由到指定的队列中去,如果存在的话。
现在我们可以指定转发器的名字了:
channel.basicPublish( "logs", "", null, message.getBytes());
Temporary queues(临时队列)
你可能还记得之前我们用队列时,会指定一个名字。队列有名字对我们来说是非常重要的——我们需要为消费者指定同一个队列。但这并不是我们的日志系统所关心的。我们要监听所有日志消息,而不仅仅是一类日志。我们只对对当前流动的消息感兴趣。解决这些问题,我需要完成两件事。
首先,每当我们连接到RabbitMQ时,需要一个新的空队列。为此我们需要创建一个随机名字的空队列,或者更好的,让服务器选好年则一个随机名字的空队列给我们。
其次,一旦消费者断开连接,队列将自动删除。
我们提供一个无参的queueDeclare()方法,创建一个非持久化、独立的、自动删除的队列,且名字是随机生成的。
String queueName = channel.queueDeclare().getQueue();
queueName是一个随机队列名。看起来会像amq.gen-JzTY20BRgKO-HjmUJj0wLg。
或者指定队列名称的非持久化,独立的,自动删除的队列:
channel.queueDeclare(QUEUE_NAME,false,true,false,null);
Bindings(绑定)
我们已经创建了一个广播的转发器和一个随机队列。现在需要告诉转发器转发消息到队列。这个关联转发器和队列的我们叫它Binding。
channel.queueBind(queueName, "logs", "");
这样,日志转发器将附加到日志队列上去。
完整的例子:
public class EmitLog { private final static String EXCHANGE_NAME="logs"; public static void main(String[] args) throws Exception{ ConnectionFactory factory = new ConnectionFactory(); //RabbitMQ所在主机ip或者主机名 factory.setHost("192.168.20.87"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); Connection connection = factory.newConnection(); Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME,"fanout"); for (int i = 0; i < 3; i++) { String message = "Hello World !"; channel.basicPublish(EXCHANGE_NAME,"",null,message.getBytes()); System.out.println(" [x] Sent '" + message + "'"); } channel.close(); connection.close(); } }
消费者1:ReceiveLogs2Console
public class ReceiveLogs2Console { private static final String EXCHANGE_NAME = "logs"; public static void main(String[] args) throws Exception{ ConnectionFactory factory = new ConnectionFactory(); factory.setHost("192.168.20.87"); factory.setPort(5672); factory.setUsername("admin"); factory.setPassword("admin"); Connection connection = factory.newConnection(); final Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME,"fanout"); //设置匿名队列,并获取队列名称 String queueName = channel.queueDeclare().getQueue(); //Exchange(交换机)与队列绑定 channel.queueBind(queueName,EXCHANGE_NAME,""); System.out.println(" [*] Waiting for messages. To exit press CTRL+C"); final 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 '" + message + "'"); } }; channel.basicConsume(queueName,true,consumer); } }
可以看到我们1个生产者用于发送log消息,2个消费者,一个用于显示,一个用于记录文件。
生产者声明了一个广播模式的转换器,订阅这个转换器的消费者都可以收到每一条消息。可以看到在生产者中,没有声明队列。这也验证了之前说的。生产者其实只关心exchange,至于exchange会把消息转发给哪些队列,并不是生产者关心的。
2个消费者,一个打印日志,一个写入文件,除了这2个地方不一样,其他地方一模一样。也是声明一下广播模式的转换器,而队列则是随机生成的,消费者实例启动后,会创建一个随机实例,这个在管理页面可以看到(如图)。而实例关闭后,随机队列也会自动删除。最后将队列与转发器绑定
注:运行的时候要先运行2个消费者实例,然后在运行生产者实例。否则获取不到实例。
看看最终的结果吧:
相关文章推荐
- 轻松搞定RabbitMQ(四)——发布/订阅
- 轻松搞定RabbitMQ(四)——发布/订阅
- 轻松搞定RabbitMQ(四)——发布/订阅
- RabbitMQ (三) 发布/订阅
- RabbitMQ学习总结(5)——发布和订阅实例详解
- RabbitMQ (三) 发布/订阅
- RabbitMQ (三) 发布/订阅
- 轻松搞定RabbitMQ
- RabbitMQ系列教程之三:发布\/订阅(Publish\/Subscribe)
- rabbitmq(三)Publish/Subscribe(发布/订阅)
- RabbitMQ学习总结 第四篇:发布/订阅 Publish/Subscribe
- 轻松搞定RabbitMQ(六)——主题
- RabbitMQ (三) 发布/订阅
- RabbitMQ学习总结(5)——发布和订阅实例详解
- RabbitMQ (三) 发布/订阅
- RabbitMQ (三) 发布/订阅
- RabbitMQ-(三) 发布/订阅
- RabbitMQ消息队列(三): 发布/订阅
- RabbitMQ官方中文入门教程(PHP版) 第三部分:发布/订阅(Publish/Subscribe)
- 今天搞定了SQL数据库同步(发布,订阅)