admin管理员组文章数量:1642244
上篇文章中,我们把每个Message都是deliver到某个Consumer。在这篇文章中,我们将会将同一个Message deliver到多个Consumer中。这个模式也被成为 “publish / subscribe”。
这篇文章中,我们将创建一个日志系统,它包含两个部分:第一个部分是发出log(Producer),第二个部分接收到并打印(Consumer)。 我们将构建两个Consumer,第一个将log写到物理磁盘上;第二个将log输出的屏幕。
一、Exchanges
RabbitMQ 的Messaging Model就是Producer并不会直接发送Message到queue。实际上,Producer并不知道它发送的Message是否已经到达queue。
Producer发送的Message实际上是发到了Exchange中。它的功能也很简单:从Producer接收Message,然后投递到queue中。Exchange需要知道如何处理Message,是把它放到那个queue中,还是放到多个queue中?这个rule是通过Exchange 的类型定义的。
我们知道有三种类型的Exchange:direct, topic 和fanout。fanout就是广播模式,会将所有的Message都放到它所知道的queue中。创建一个名字为logs,类型为fanout的Exchange:
channel.exchangeDeclare("logs", "fanout");
Listing exchanges
通过rabbitmqctl可以列出当前所有的Exchange:
sudo rabbitmqctl list_exchanges
注意 amq.* exchanges 和the default (unnamed)exchange是RabbitMQ默认创建的。
默认exchanges
上几节我们通过“”来指定默认的exchanges帮助我们分发消息给queue,像下面这样:
channel.basicPublish("", "hello", null, message.getBytes());
默认的分发机制是通过routingKey来匹配queue
现在我们可以通过exchange,而不是routing_key来publish Message了:
channel.basicPublish( "logs", "", null, message.getBytes());
二、Temporary queues(临时queues)
截至现在,我们用的queue都是有名字的:第一个是hello,第二个是task_queue。使用有名字的queue,使得在Producer和Consumer之前共享queue成为可能。
但是对于我们将要构建的日志系统,并不需要有名字的queue。我们希望得到所有的log,而不是它们中间的一部分。而且我们只对当前的log感兴趣。为了实现这个目标,我们需要两件事情:
- 每当Consumer连接时,我们需要一个新的,空的queue。因为我们不对老的log感兴趣。幸运的是,如果在声明queue时不指定名字,那么RabbitMQ会随机为我们选择这个名字。方法:
- 当Consumer关闭连接时,这个queue要被deleted。
String queueName = channel.queueDeclare().getQueue();
这样我们就能得到一个新的,自动删除的,随机分配名字的queue
三、Bindings(绑定)
现在我们已经创建了fanout类型的exchange和没有名字的queue(实际上是RabbitMQ帮我们取了名字)。那exchange怎么样知道它的Message发送到哪个queue呢?答案就是通过bindings:绑定。
channel.queueBind(queueName, "logs", "");
现在logs的exchange就将它的Message附加到我们创建的queue了
Listing bindings
你可以通过以下命令来获取当前已经存在的binding
rabbitmqctl list_bindings
四、最终版本
Producer,在这里就是产生log的program,基本上和前几个都差不多。最主要的区别就是publish通过了exchange而不是routing_key。
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。publish到一个不存在的exchange是被禁止的。如果没有queue bindings exchange的话,log是被丢弃的。
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);
}
}
本文标签: 基础教程RabbitMQSubscribepublish
版权声明:本文标题:RabbitMQ 基础教程PublishSubscribe(发布订阅) 内容由热心网友自发贡献,该文观点仅代表作者本人, 转载请联系作者并注明出处:https://m.elefans.com/xitong/1729333443a1196685.html, 本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌抄袭侵权/违法违规的内容,一经查实,本站将立刻删除。
发表评论