跳至主内容

RabbitMQ 教程 - 发布/订阅

发布/订阅​

(使用 Spring AMQP)​

信息

先决条件​

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

哪里寻求帮助​

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

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

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

在本部分中，我们将实现 fanout 模式，将消息传递给多个消费者。这种模式也称为“发布/订阅”，通过在我们的 Tut3Config 文件中配置多个 bean 来实现。

本质上，发布的消息将被广播给所有接收者。

交换器​

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

让我们快速回顾一下之前的内容

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

RabbitMQ 消息模型的核心思想是生产者从不直接将任何消息发送到队列。实际上，生产者往往甚至不知道消息是否会被传递到任何队列。

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

有几种可用的交换机类型：direct、topic、headers 和 fanout。我们将重点介绍最后一个——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）。它们都是运行 fanout 配置文件教程的同义词。接下来，我们将 FanoutExchange 配置为 Spring bean。在 Tut3Receiver 类中，我们定义了四个 bean：2 个 AnonymousQueue（在 AMQP 术语中为非持久化、独占、自动删除队列）和 2 个绑定，用于将这些队列绑定到交换机。

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

列出交换器​

要列出服务器上的交换器，可以使用一直很有用的 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 吗）。能够命名队列对我们很重要——我们需要将工作者指向同一个队列。当您想在生产者和消费者之间共享队列时，为队列命名很重要。

但我们的 fanout 示例并非如此。我们想收听所有消息，而不仅仅是一部分。我们只对当前流的消息感兴趣，而不是旧的消息。要解决这个问题，我们需要两样东西。

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

其次，一旦我们断开消费者连接，队列应该自动删除。要使用 Spring AMQP 客户端做到这一点，我们定义了一个 AnonymousQueue，它创建了一个具有生成名称的非持久化、独占、自动删除队列。

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

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

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

绑定​

我们已经创建了一个 fanout 交换机和一个队列。现在我们需要告诉交换机将消息发送到我们的队列。交换机和队列之间的这种关系称为绑定。在上面的 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 源代码

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

./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。

© . This site is unofficial and not affiliated with VMware.