跳至主要内容

RabbitMQ 教程 - 发布/订阅

发布/订阅​

(使用 Spring AMQP)​

信息

先决条件​

本教程假设 RabbitMQ 已安装并在 localhost 上的标准端口 (5672) 上运行。如果您使用不同的主机、端口或凭据，则连接设置需要调整。

获取帮助​

如果您在学习本教程时遇到问题，可以通过GitHub 讨论区或RabbitMQ 社区 Discord联系我们。

在第一个教程中，我们展示了如何使用 start.spring.io 利用 Spring Initializr 创建一个包含 RabbitMQ 启动器依赖项的项目，以创建 Spring AMQP 应用程序。

在上一个教程中，我们创建了一个新包 tut2 来放置我们的配置、发送者和接收者，并创建了一个有两个消费者的工作队列。工作队列背后的假设是每个任务都只交付给一个工作者。

在本部分中，我们将实现扇出模式，以将消息传递给多个消费者。此模式也称为“发布/订阅”，它是通过在我们的 Tut3Config 文件中配置多个 Bean 来实现的。

从本质上讲，发布的消息将广播到所有接收者。

交换机​

在教程的前一部分中，我们向队列发送和接收消息。现在是时候介绍 RabbitMQ 中的完整消息模型了。

让我们快速回顾一下我们在之前的教程中介绍的内容

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

RabbitMQ 中消息模型的核心思想是生产者永远不会直接向队列发送任何消息。实际上，很多时候生产者甚至不知道消息是否会被传递到任何队列。

相反，生产者只能将消息发送到交换机。交换机是一个非常简单的东西。一方面，它从生产者那里接收消息；另一方面，它将消息推送到队列。交换机必须准确地知道如何处理它收到的消息。应该将其追加到特定队列吗？应该追加到多个队列吗？或者应该丢弃它？这些规则由交换机类型定义。

有几种可用的交换机类型：direct、topic、headers 和 fanout。我们将重点关注最后一个——扇出。让我们配置一个 Bean 来描述这种类型的交换机，并将其命名为 tut.fanout

import org.springframework.amqp.core.*;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Profile;


@Profile({"tut3", "pub-sub", "publish-subscribe"})
@Configuration
public class Tut3Config {

	@Bean
	public FanoutExchange fanout() {
		return new FanoutExchange("tut.fanout");
	}

	@Profile("receiver")
	private static class ReceiverConfig {

		@Bean
		public Queue autoDeleteQueue1() {
			return new AnonymousQueue();
		}

		@Bean
		public Queue autoDeleteQueue2() {
			return new AnonymousQueue();
		}

		@Bean
		public Binding binding1(FanoutExchange fanout,
		    Queue autoDeleteQueue1) {
			return BindingBuilder.bind(autoDeleteQueue1).to(fanout);
		}

		@Bean
		public Binding binding2(FanoutExchange fanout,
		    Queue autoDeleteQueue2) {
			return BindingBuilder.bind(autoDeleteQueue2).to(fanout);
		}

		@Bean
		public Tut3Receiver receiver() {
			return new Tut3Receiver();
		}
	}

	@Profile("sender")
	@Bean
	public Tut3Sender sender() {
		return new Tut3Sender();
	}
}

我们遵循与前两个教程相同的方法。我们为教程创建了三个配置文件（tut3、pub-sub 或 publish-subscribe）。它们都是运行扇出配置文件教程的同义词。接下来，我们将 FanoutExchange 配置为 Spring Bean。在 Tut3Receiver 类中，我们定义了四个 Bean：2 个 AnonymousQueue（在 AMQP 术语中是非持久的、排他的、自动删除的队列）和 2 个绑定，用于将这些队列绑定到交换机。

扇出交换机非常简单。正如您可能从名称中猜到的那样，它只是将接收到的所有消息广播到它知道的所有队列。这正是我们需要将消息扇出的方法。

列出交换机​

要列出服务器上的交换机，您可以运行非常有用的 rabbitmqctl

sudo rabbitmqctl list_exchanges

在此列表中，将有一些 amq.* 交换机和默认（未命名）交换机。这些是默认创建的，但您现在不太可能需要使用它们。

无名交换机​

在教程的前一部分中，我们对交换机一无所知，但仍然能够向队列发送消息。这是因为我们使用了默认交换机，我们通过空字符串 ("") 来标识它。

回想一下我们之前是如何发布消息的

   template.convertAndSend(queue.getName(), message)

第一个参数是路由键，RabbitTemplate 默认将消息发送到默认交换机。每个队列都自动绑定到默认交换机，队列名称作为绑定键。这就是为什么我们可以使用队列名称作为路由键来确保消息最终进入队列。

现在，我们可以改为发布到我们命名的交换机

@Autowired
private RabbitTemplate template;

@Autowired
private FanoutExchange fanout;   // configured in Tut3Config above

template.convertAndSend(fanout.getName(), "", message);

从现在开始，fanout 交换机将消息追加到我们的队列。

临时队列​

您可能还记得，我们之前使用的是具有特定名称的队列（记住 hello）。能够命名队列对我们来说至关重要——我们需要将工作者指向同一个队列。当您希望在生产者和消费者之间共享队列时，为队列命名非常重要。

但这不适用于我们的扇出示例。我们希望了解所有消息，而不仅仅是其中的一部分。我们也只对当前流动的消息感兴趣，而不是旧消息。为了解决这个问题，我们需要两件事。

首先，每当我们连接到 Rabbit 时，都需要一个新的空队列。为此，我们可以创建一个具有随机名称的队列，或者——甚至更好——让服务器为我们选择一个随机的队列名称。

其次，一旦我们断开消费者的连接，队列应该自动删除。要使用 Spring AMQP 客户端执行此操作，我们定义了一个 AnonymousQueue，它会创建一个具有生成的名称的非持久性、排他性、自动删除的队列

@Bean
public Queue autoDeleteQueue1() {
	return new AnonymousQueue();
}

@Bean
public Queue autoDeleteQueue2() {
	return new AnonymousQueue();
}

此时，我们的队列具有随机队列名称。例如，它可能看起来像 spring.gen-1Rx9HOqvTAaHeeZrQWu8Pg。

绑定​

我们已经创建了一个扇出交换机和一个队列。现在我们需要告诉交换机将消息发送到我们的队列。交换机和队列之间的这种关系称为绑定。在上面的 Tut3Config 中，您可以看到我们有两个绑定，每个 AnonymousQueue 一个。

@Bean
public Binding binding1(FanoutExchange fanout,
        Queue autoDeleteQueue1) {
	return BindingBuilder.bind(autoDeleteQueue1).to(fanout);
}

列出绑定​

您可以使用（您猜对了）列出现有的绑定，

rabbitmqctl list_bindings

综合起来​

发出消息的生产者程序与之前的教程没有太大区别。最重要的变化是我们现在希望将消息发布到我们的 fanout 交换机而不是无名交换机。发送时需要提供 routingKey，但对于 fanout 交换机，其值会被忽略。以下是 tut3.Sender.java 程序的代码

package org.springframework.amqp.tutorials.tut3;

import org.springframework.amqp.core.FanoutExchange;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import java.util.concurrent.atomic.AtomicInteger;

public class Tut3Sender {

	@Autowired
	private RabbitTemplate template;

	@Autowired
	private FanoutExchange fanout;

	AtomicInteger dots = new AtomicInteger(0);

	AtomicInteger count = new AtomicInteger(0);

	@Scheduled(fixedDelay = 1000, initialDelay = 500)
	public void send() {
		StringBuilder builder = new StringBuilder("Hello");
		if (dots.getAndIncrement() == 3) {
			dots.set(1);
		}
		for (int i = 0; i < dots.get(); i++) {
			builder.append('.');
		}
		builder.append(count.incrementAndGet());
		String message = builder.toString();
		template.convertAndSend(fanout.getName(), "", message);
		System.out.println(" [x] Sent '" + message + "'");
	}

}

Tut3Sender.java 源代码

如您所见，我们利用了 Tut3Config 文件中的 Bean，并自动装配了 RabbitTemplate 以及我们配置的 FanoutExchange。此步骤是必要的，因为禁止发布到不存在的交换机。

如果尚无队列绑定到交换机，则消息将丢失，但对我们来说没关系；如果尚无消费者在侦听，我们可以安全地丢弃消息。

Tut3Receiver.java 的代码

import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.util.StopWatch;

public class Tut3Receiver {

	@RabbitListener(queues = "#{autoDeleteQueue1.name}")
	public void receive1(String in) throws InterruptedException {
		receive(in, 1);
	}

	@RabbitListener(queues = "#{autoDeleteQueue2.name}")
	public void receive2(String in) throws InterruptedException {
		receive(in, 2);
	}

	public void receive(String in, int receiver) throws InterruptedException {
		StopWatch watch = new StopWatch();
		watch.start();
		System.out.println("instance " + receiver + " [x] Received '" + in + "'");
		doWork(in);
		watch.stop();
		System.out.println("instance " + receiver + " [x] Done in "
		    + watch.getTotalTimeSeconds() + "s");
	}

	private void doWork(String in) throws InterruptedException {
		for (char ch : in.toCharArray()) {
			if (ch == '.') {
				Thread.sleep(1000);
			}
		}
	}

}

Tut3Receiver.java 源代码

像以前一样编译，我们就可以执行扇出发送者和接收者了。

./mvnw clean package

当然，要执行教程，请执行以下操作

# shell 1
java -jar target/rabbitmq-tutorials.jar --spring.profiles.active=pub-sub,receiver \
    --tutorial.client.duration=60000
# shell 2
java -jar target/rabbitmq-tutorials.jar --spring.profiles.active=pub-sub,sender \
    --tutorial.client.duration=60000

使用 rabbitmqctl list_bindings，您可以验证代码是否按我们想要的方式创建了绑定和队列。如果运行了两个 ReceiveLogs.java 程序，您应该会看到类似以下内容

sudo rabbitmqctl list_bindings
tut.fanout	exchange	8b289c9c-a1eb-4a3a-b6a9-163c4fdcb6c2	queue		[]
tut.fanout	exchange	d7e7d193-65b1-4128-a532-466a5256fd31	queue		[]

结果的解释很简单：来自交换机 logs 的数据进入两个具有服务器分配名称的队列。这正是我们想要的结果。

要了解如何侦听消息的子集，让我们继续学习教程 4

© 2024 RabbitMQ. All rights reserved.