跳至主内容

RabbitMQ 教程 - 路由

路由​

(使用 Spring AMQP)​

信息

先决条件​

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

哪里寻求帮助​

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

在上一篇教程中，我们构建了一个简单的扇出交换器。我们能够将消息广播给多个接收者。

在本教程中，我们将为其添加一项功能——我们将能够仅订阅消息的子集。例如，我们将能够仅将消息定向到感兴趣的特定颜色（“橙色”、“黑色”、“绿色”），同时仍然能够在控制台上打印所有消息。

绑定​

在之前的示例中，我们已经创建了绑定。您可能还记得我们 Tut3Config 文件中的类似代码

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

绑定是交换机和队列之间的关系。这可以简单地理解为：队列对来自此交换机的消息感兴趣。

绑定可以接受一个额外的绑定键参数。Spring AMQP 使用流畅的 API 来使这种关系非常清晰。我们将交换器和队列传递给 BindingBuilder，然后简单地将队列“绑定”到交换器“使用绑定键”，如下所示

@Bean
public Binding binding1a(DirectExchange direct,
    Queue autoDeleteQueue1) {
    return BindingBuilder.bind(autoDeleteQueue1)
        .to(direct)
        .with("orange");
}

绑定 key 的含义取决于交换机类型。我们之前使用的 fanout 交换机只是忽略了它的值。

直连交换机​

我们上一篇教程中的消息传递系统会将所有消息广播给所有消费者。我们希望将其扩展为允许根据消息的颜色类型过滤消息。例如，我们可能希望一个将日志消息写入磁盘的程序只接收关键错误，而不会在警告或信息日志消息上浪费磁盘空间。

我们使用的是fanout交换，这并没有给我们太大的灵活性——它只能进行无差别的广播。

我们将改用 direct 交换器。direct 交换器背后的路由算法很简单——消息会发送到绑定键与消息路由键完全匹配的队列。

为了说明这一点，请考虑以下设置

在此设置中，我们可以看到 direct 交换机 X 绑定了两个队列。第一个队列绑定了绑定 key orange，第二个队列有两个绑定，一个绑定 key 是 black，另一个是 green。

在这种设置下，发布到具有路由 key orange 的交换机的消息将被路由到队列 Q1。具有路由 key black 或 green 的消息将发送到 Q2。所有其他消息将被丢弃。

多个绑定​

使用相同的绑定 key 绑定多个队列是完全合法的。在我们的示例中，我们可以添加一个 X 和 Q1 之间的绑定，绑定 key 为 black。在这种情况下，direct 交换机将表现得像 fanout，并将消息广播给所有匹配的队列。路由 key 为 black 的消息将被传递给 Q1 和 Q2。

发布消息​

我们将为此路由系统使用此模型。我们将消息发送到一个 direct 交换器，而不是 fanout。我们将颜色作为路由键提供。这样，接收程序将能够选择它想要接收（或订阅）的颜色。让我们先专注于发送消息。

一如既往，我们在 Tut4Config 中进行一些 Spring Boot 配置

@Bean
public DirectExchange direct() {
    return new DirectExchange("tut.direct");
}

我们已经准备好发送消息了。如图所示，颜色可以是“橙色”、“黑色”或“绿色”之一。

订阅​

接收消息的操作将与上一篇教程类似，但有一个例外——我们将为我们感兴趣的每种颜色创建一个新的绑定。这也会包含在 Tut4Config 中

@Bean
public DirectExchange direct() {
    return new DirectExchange("tut.direct");
}
...
@Bean
public Binding binding1a(DirectExchange direct,
    Queue autoDeleteQueue1) {
    return BindingBuilder.bind(autoDeleteQueue1)
        .to(direct)
        .with("orange");
}

总而言之​

与之前的教程一样，为本教程创建一个名为 tut4 的新包，并创建 Tut4Config 类。Tut4Config.java 类的代码

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

@Profile({"tut4","routing"})
@Configuration
public class Tut4Config {

	@Bean
	public DirectExchange direct() {
		return new DirectExchange("tut.direct");
	}

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

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

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

		@Bean
		public Binding binding1a(DirectExchange direct,
		    Queue autoDeleteQueue1) {
			return BindingBuilder.bind(autoDeleteQueue1)
			    .to(direct)
			    .with("orange");
		}

		@Bean
		public Binding binding1b(DirectExchange direct,
		    Queue autoDeleteQueue1) {
			return BindingBuilder.bind(autoDeleteQueue1)
			    .to(direct)
			    .with("black");
		}

		@Bean
		public Binding binding2a(DirectExchange direct,
		    Queue autoDeleteQueue2) {
			return BindingBuilder.bind(autoDeleteQueue2)
			    .to(direct)
			    .with("green");
		}

		@Bean
		public Binding binding2b(DirectExchange direct,
		    Queue autoDeleteQueue2) {
			return BindingBuilder.bind(autoDeleteQueue2)
			    .to(direct)
			    .with("black");
		}

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

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

我们的发送方类的代码是

package org.springframework.amqp.tutorials.tut4;

import org.springframework.amqp.core.DirectExchange;
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 Tut4Sender {

	@Autowired
	private RabbitTemplate template;

	@Autowired
	private DirectExchange direct;

	AtomicInteger index = new AtomicInteger(0);

	AtomicInteger count = new AtomicInteger(0);

	private final String[] keys = {"orange", "black", "green"};

	@Scheduled(fixedDelay = 1000, initialDelay = 500)
	public void send() {
		StringBuilder builder = new StringBuilder("Hello to ");
		if (this.index.incrementAndGet() == 3) {
			this.index.set(0);
		}
		String key = keys[this.index.get()];
		builder.append(key).append(' ');
		builder.append(this.count.get());
		String message = builder.toString();
		template.convertAndSend(direct.getName(), key, message);
		System.out.println(" [x] Sent '" + message + "'");
	}

}

Tut4Receiver.java 的代码是

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

public class Tut4Receiver {

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

}

像往常一样编译（请参阅 教程一 关于 Maven 编译和从 jar 执行选项）。

./mvnw clean package

在一个终端窗口中，您可以运行

java -jar target/rabbitmq-tutorials.jar \
    --spring.profiles.active=routing,receiver \
    --tutorial.client.duration=60000

在另一个终端窗口中运行发送方

java -jar target/rabbitmq-tutorials.jar \
    --spring.profiles.active=routing,sender \
    --tutorial.client.duration=60000

完整源代码：Tut4Receiver.java 源代码 和 Tut4Sender.java 源代码。配置在 Tut4Config.java 源代码 中。

继续学习 教程 5，了解如何根据模式监听消息。

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