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

时间：2022-01-03 19:22:02

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

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

交换机(Exchanges)

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

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

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

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

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

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

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

前面发布消息的方式：

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

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

临时队列(Temporary queues)

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

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

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

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

绑定(Bindings)

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();
        }
    }

}
标签：

相关文章