RabbitMQ入门(3)——发布/订阅(Publish/Subscribe)

时间:2022-01-13 16:48:20

在上一篇RabbitMQ入门(2)——工作队列中,有一个默认的前提:每个任务都只发送到一个工作人员。这一篇将介绍发送一个消息到多个消费者。这种模式称为发布/订阅(Publish/Subscribe)

为了说明这种模式,我们将会构建一个简单的日志系统。它包含三个程序:生成日志消息的程序,将消息打印到控制台的程序和将消息保存到日志文件的程序。

交换机(Exchange)

前面介绍了队列如何接发消息。接下来介绍RabbitMQ的完整消息传递模型。首先快速回顾一下前一篇的内容:

  • 生产者是发送消息的用户应用程序
  • 队列是存储消息的缓存
  • 消费者是接收消息的用户应用程序

RabbitMQ的消息传递模型核心思想是生产者从不将任何消息直接发送到队列。实际上,通常来说,生产者甚至不知道消息是否会被发送到任何队列。相反,生产者只能发送消息到交换机。一方面,交换机从生产者接收消息,另一方面,将消息推送到队列中。交换机必须知道如何处理接收到的消息。是否应该添加到具体队列里?还是添加到多个队列里?还是应该被放弃。这些规则是由交换机类型决定的。交换机类型:directtopicheadersfanout

RabbitMQ入门(3)——发布/订阅(Publish/Subscribe)

下面我们将首先介绍fanout类型。fanout,顾名思义,它的方式就是将接收到的所有消息都发送到它所知道的队列中。声明交换机:

channel.exchangeDeclare("test_logs", "fanout");

在前面的例子中,并没有提到交换机,但是依旧能将消息发送到队列。这是因为使用了一个默认的交换机,它的标识符是""

前面发布消息的方式:

channel.basicPublish("", "hello", null, message.getBytes());

第一个参数就是交换机的名称。空字符串表示默认值或匿名交换机,如果routingKey存在(第二个参数),消息由routingKey决定发送到哪个队列。

临时队列(Temporary queue)

前面的例子中,队列都有一个指定的名称,因为它对于在生产者和消费者之间共享队列是至关重要的。然而,在本篇的日志系统中并不重要,我们需要了解的是所有的日志消息,而不是其中的一部分,同时,我们只对当前正在传递的消息感兴趣。为了满足这种需要,需要满足两点:

  • 当连接到RabbitMQ时,需要一个新的,空的队列。可以随机生成一个队列名。或者更好地,让RabbitMQ为我们选择一个随机的队列名。
  • 当消费者断开连接时,队列需要自动删除。

创建一个非持久化的、唯一的、自动删除的队列:

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

绑定(Binding)

RabbitMQ入门(3)——发布/订阅(Publish/Subscribe)

现在,我们已经创建了一个fanout交换机和队列。通过绑定告诉交换机发送消息到队列。

channel.queueBind(queue, "test_logs", "");

代码清单

日志发送端:

package com.xxyh.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeoutException; public class EmitLog {
private static final String EXCHANGE_NAME = "test_logs"; public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT); SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
String message = Thread.currentThread().getName() + " -- " + format.format(new Date()) + " logging...";
channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
System.out.println(Thread.currentThread().getName() + " 发送消息:" + message); channel.close();
connection.close();
}
}

从上面的例子可以看出,当创建一个连接后,我们声明了交换机,这一步是必须的,禁止向一个不存在的交换机发布消息。

如果没有队列绑定到交换机,消息将会丢失,这对我们来说是可行的。如果没有消费者监听,我们可以安全地丢弃这个消息。

打印日志到控制台:

package com.xxyh.rabbitmq;

import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException; public class ReceivLogsToConsole { private static final String EXCHANGE_NAME = "test_logs"; public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
String queue = channel.queueDeclare().getQueue();
channel.queueBind(queue, EXCHANGE_NAME, ""); System.out.println("准备接收消息......"); 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(Thread.currentThread().getName() + " 接收消息:" + message);
}
}; channel.basicConsume(queue, true, consumer);
} }

存储日志到文件:

package com.xxyh.rabbitmq;

import com.rabbitmq.client.*;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.concurrent.TimeoutException; public class ReceiveLogsToSave {
private static final String EXCHANGE_NAME = "test_logs"; public static void main(String[] args) throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel(); channel.exchangeDeclare(EXCHANGE_NAME, BuiltinExchangeType.FANOUT);
String queue = channel.queueDeclare().getQueue();
channel.queueBind(queue, EXCHANGE_NAME, ""); System.out.println("准备接收消息......"); 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"); print2File(message);
} };
channel.basicConsume(queue, true, consumer);
} private static void print2File(String message) throws FileNotFoundException {
try {
String dir = ReceiveLogsToSave.class.getClassLoader().getResource("").getPath();
String logFileName = new SimpleDateFormat("yyyy-MM-dd").format(new Date());
File file = new File(dir, logFileName + ".txt");
FileOutputStream fos = new FileOutputStream(file, true);
fos.flush();
fos.write((message + "\r\n").getBytes());
} catch (FileNotFoundException e) {
e.printStackTrace();
} catch (IOException e) {
e.printStackTrace();
}
} }