1.
2.
3.
在上一节中我们创建了一个工作队列,最好的情况是工作队列能够把任务恰到好处的分配给每一个 worker。这一节中我们将做一些完全不同的事情——将消息传递给每一个消费者,这种模式被称为发布 / 订阅。
为了说明这种模式,我们构建一个日志系统,其包括两个部分,一个发出日志消息,另一个接收并打印出来。
在我们的日志系统里,每一个拷贝的接收者都将收到消息,这样我们就能够跑一个接收者将收到的日志消息写入磁盘,并在同时另外一个接收者将日志消息打印在屏幕上。
本质上来说,发布日志消息将要被广播到所有的接收器。
在本教程之前我们在队列中发送和接收消息,现在我们将介绍 Rabbit 中完整的消息传递模式。
我们先回顾一下前面的教程:
生产者发送消息
队列是一个缓冲区存储消息
消费者接收消息
RabbitMQ 消息模式的核心是生产者从来不会直接发送一个消息给队列,事实上,发送者都不知道消息将要被发送到一个队列。
相反,生产者发送消息到 exchange,exchange 是一个非常简单的东西,它一方面接收生产者的消息,另一方面推送消息到队列,exchange 必须知道如何处理它接收到的消息,是应该将消息附加至一个特定的队列?还是附加到很多队列?或者而应该丢弃?这些规则由 exchange 来决定。
- channel.exchangeDeclare("logs", "fanout");
fanout 类型的 exchange 非常简单,我们可以从名字就猜出来,它只是广播接收到的所有消息至所有的队列。而这正是我们的日志系统所需要的。
列出 exchange
我们在服务器上将所有的 exchange 列出来
$ sudo rabbitmqctl list_exchanges
Listing exchanges ...
direct
amq.direct direct
amq.fanout fanout
amq.headers headers
amq.match headers
amq.rabbitmq.log topic
amq.rabbitmq.trace topic
amq.topic topic
logs fanout
...done.
在这些 list 里面我们可看到一些 amq.* exchanges,这些是默认的 exchange。
无名的 exchange
在上一节中,我们对于 exchange 什么都不知道,但是我们依旧可以发送消息至队列,这是因为我们使用了使用了默认的 exchange,它被定义为一个空字符串 ""。
channel.basicPublish("","hello", null, message.getBytes());
第一个参数就是 exchange,空字符串表示默认或者无名交换。如果 routingkey 存在的话,消息会根据 routingkey 被路由到队列。
现在我们发布到有名的 exchange:
- channel.basicPublish("logs", "", null, message.getBytes());
我们在之前的教程中给队列起了名字(helll 和 task_queue), 给一个队列命名对于我们来说是很重要的,我们需要消费者指向同一个队列。
但是这不是我们目前的日志程序所需要的,我们希望能接受所有的消息,而不是这些消息的子集。当然我们也只是关注当前流动的消息,而不需要关注那些过期的消息。这样,我们需要做两件事情。
第一,无论何时我们连接 Rabbit,我们需要一个新的空的队列,这样需要用随机的名字来创建一个队列,更好地,我们让服务器来随机的生成一个队列名字。
第二,我们一旦和消费者断开,队列必须马上删除。
在 Java Client 中,我们不提供参数给 queueDeclare() 方法,就能生成一个非持久,独特,自动删除队列的名称:
- String queueName = channel.queueDeclare().getQueue();
这样队列拥有一个随机的队列名称,比如:amq.gen-JzTY20BRgKO-HjmUJj0wLg。
现在我们已经建立了一个 exchange 和一个队列,现在我们需要告诉 exchange 将消息发送到我们的队列中,exchange 和队列之间的关系我们成为 Binding。
- channel.queueBind(queueName, "logs", "");
从现在开始日志程序中的 exchange 就可以将消息附加到队列上了。
列出 binding 的列表
你可以列出现有的 Binding 列表,命令是:rabbitmqctl list_bindings.
生产者和之前的教程中相差不多,最大的改变是我们现在需要将消息发布到 logs exchange,而不是之前的未命名 exchange,当发送消息的时候我们需要提供一个 routingKey,但是在 fanout 的消息模式中被忽略了。
EmitLog.java:
- import java.io.IOException;
- import com.rabbitmq.client.ConnectionFactory;
- import com.rabbitmq.client.Connection;
- import com.rabbitmq.client.Channel;
- public class EmitLog {
- private static final String EXCHANGE_NAME = "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, "fanout");
- String message = getMessage(argv);
- channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
- System.out.println(" [x] Sent '" + message + "'");
- channel.close();
- connection.close();
- }
- //...
- }
建立连接后,我们定义了一个 exchange,因为发布到一个不存在的 exchange 是被禁止的。
如果没有队列绑定到 exchange,信息会丢失,但是也没关系。如果没有消费者在监听,我们可以安全的丢弃消息。
ReceiveLogs.java:
- import com.rabbitmq.client. * ;
- import java.io.IOException;
- public class ReceiveLogs {
- private static final String EXCHANGE_NAME = "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, "fanout");
- String queueName = channel.queueDeclare().getQueue();
- channel.queueBind(queueName, EXCHANGE_NAME, "");
- 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 '" + message + "'");
- }
- };
- channel.basicConsume(queueName, true, consumer);
- }
- }
来源: http://www.cnblogs.com/aheizi/p/5782469.html