跳至主内容

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

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

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

列出交换器​

要列出服务器上的交换器，可以使用一直很有用的 rabbitmqctl。

sudo rabbitmqctl list_exchanges

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

无名交换机 (Nameless exchange)​

在之前的教程中，我们对交换器一无所知，但仍然能够将消息发送到队列。之所以可能，是因为我们使用了默认交换器，我们用空字符串 ("") 来标识它。

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

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

第一个参数是路由键（routing key），而 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 交换机和一个队列。现在我们需要告诉交换机将消息发送到我们的队列。交换机和队列之间的这种关系称为 绑定 (binding)。在上面的 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，并使用自动装配（autowire）注入了 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.