跳至主内容

RabbitMQ 教程 - 工作队列

工作队列​

(使用 Spring AMQP)​

信息

先决条件​

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

哪里寻求帮助​

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

在第一个教程中，我们编写了程序来发送和接收来自命名队列的消息。在这个教程中，我们将创建一个工作队列，用于在多个工作者之间分发耗时的任务。

工作队列（又称：任务队列）背后的主要思想是避免立即执行一个资源密集型任务并等待其完成。相反，我们安排任务稍后执行。我们将一个任务封装成一条消息并将其发送到队列。一个在后台运行的工作进程会接收任务并最终执行工作。当您运行多个工作进程时，任务将在它们之间共享。

这个概念对于 Web 应用程序尤其有用，在这些应用程序中，不可能在短暂的 HTTP 请求窗口内处理一个复杂的任务。

准备工作​

在本教程的前一部分，我们发送了一条包含“Hello World!”的消息。现在我们将发送代表复杂任务的字符串。我们没有实际的现实世界任务，比如需要调整大小的图片或需要渲染的PDF文件，所以我们通过使用Thread.sleep()函数来模拟工作——假装我们很忙。我们将字符串中的点数作为其复杂度的度量；每一点将代表一秒钟的“工作”。例如，一个由Hello...描述的假任务将需要三秒钟。

如果您还没有设置项目，请参阅第一个教程中的设置。我们将遵循第一个教程相同的模式：1) 创建一个名为tut2的包，并创建Tut2Config、Tut2Receiver和Tut2Sender类。首先，创建一个名为tut2的新包，我们将在此放置三个类。在配置类中，我们设置了两个配置文件，教程的标签为tut2，模式的名称为work-queues。我们利用Spring将队列暴露为一个Bean。我们将接收器设置为一个配置文件，并定义两个Bean来对应上面图中的工作者；receiver1和receiver2。最后，我们为发送者定义一个配置文件并定义发送者Bean。配置现在就完成了。

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

@Profile({"tut2", "work-queues"})
@Configuration
public class Tut2Config {

    @Bean
    public Queue hello() {
        return new Queue("hello");
    }

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

        @Bean
        public Tut2Receiver receiver1() {
            return new Tut2Receiver(1);
        }

        @Bean
        public Tut2Receiver receiver2() {
            return new Tut2Receiver(2);
        }
    }

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

发送者​

我们将修改发送者，通过一种非常随意的方式在消息末尾追加一个点来提供一种识别任务是否为长时间运行任务的方法，使用RabbitTemplate的相同方法来发布消息，即convertAndSend。文档将其定义为，“将Java对象转换为消息，并将其发送到具有默认路由键的默认交换机。”

package org.springframework.amqp.tutorials.tut2;

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

	@Autowired
	private RabbitTemplate template;

	@Autowired
	private Queue queue;

	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.incrementAndGet() == 4) {
			dots.set(1);
		}
		for (int i = 0; i < dots.get(); i++) {
			builder.append('.');
		}
		builder.append(count.incrementAndGet());
		String message = builder.toString();
		template.convertAndSend(queue.getName(), message);
		System.out.println(" [x] Sent '" + message + "'");
	}

}

接收者​

我们的接收者Tut2Receiver在doWork()方法中模拟了假任务的任意长度，其中点的数量转化为工作所需秒数。同样，我们利用@RabbitListener监听hello队列，并使用@RabbitHandler接收消息。正在处理消息的实例被添加到我们的监视器中，以显示哪个实例、消息以及处理消息所需的时间。

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

@RabbitListener(queues = "hello")
public class Tut2Receiver {

    private final int instance;

    public Tut2Receiver(int i) {
        this.instance = i;
    }

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

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

总而言之​

使用mvn package编译它们，然后使用以下选项运行

./mvnw clean package

# shell 1
java -jar target/rabbitmq-tutorials.jar --spring.profiles.active=work-queues,receiver
# shell 2
java -jar target/rabbitmq-tutorials.jar --spring.profiles.active=work-queues,sender

发送者的输出应该类似于

Ready ... running for 10000ms
 [x] Sent 'Hello.1'
 [x] Sent 'Hello..2'
 [x] Sent 'Hello...3'
 [x] Sent 'Hello.4'
 [x] Sent 'Hello..5'
 [x] Sent 'Hello...6'
 [x] Sent 'Hello.7'
 [x] Sent 'Hello..8'
 [x] Sent 'Hello...9'
 [x] Sent 'Hello.10'

以及工作者的输出应该类似于

Ready ... running for 10000ms
instance 1 [x] Received 'Hello.1'
instance 2 [x] Received 'Hello..2'
instance 1 [x] Done in 1.001s
instance 1 [x] Received 'Hello...3'
instance 2 [x] Done in 2.004s
instance 2 [x] Received 'Hello.4'
instance 2 [x] Done in 1.0s
instance 2 [x] Received 'Hello..5'

消息确认​

执行任务可能需要几秒钟。您可能会想，如果一个消费者开始了一个耗时任务，然后在部分完成时死亡，会发生什么？Spring AMQP默认对消息确认采取保守的策略。如果侦听器抛出异常，容器将调用

channel.basicReject(deliveryTag, requeue)

默认情况下，重新排队为true，除非您显式设置

defaultRequeueRejected=false

或者侦听器抛出AmqpRejectAndDontRequeueException。这通常是您希望从侦听器处获得的行为。在此模式下，无需担心遗忘确认。处理完消息后，侦听器将调用

channel.basicAck()

确认消息必须在接收到传输的同一通道上发送。尝试使用不同的通道进行确认将导致通道级别的协议异常。请参阅确认文档指南以了解更多信息。Spring AMQP通常会处理这些，但在与直接使用RabbitMQ Java客户端的代码结合使用时，需要注意这一点。

遗忘的确认​

错过basicAck是一个常见的错误，Spring AMQP通过其默认配置帮助避免了这种情况。其后果是严重的。当您的客户端退出时，消息将被重新传递（这可能看起来像是随机的重新传递），但RabbitMQ将消耗越来越多的内存，因为它无法释放任何未确认的消息。

为了调试此类错误，您可以使用 rabbitmqctl 打印 messages_unacknowledged 字段

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

在 Windows 上，删除 sudo

rabbitmqctl.bat list_queues name messages_ready messages_unacknowledged

消息持久化​

默认情况下，Spring AMQP的消息是持久化的。请注意，消息最终会进入的队列也需要是持久化的，否则，当代理重新启动时，消息不会被保留，因为非持久化队列本身也不会在重启后被保留。

要对消息持久化或出站消息的各个方面进行更多控制，您需要使用接受MessagePostProcessor参数的RabbitTemplate#convertAndSend(...)方法。MessagePostProcessor在消息实际发送之前提供一个回调，因此这是一个修改消息负载或标头的好地方。

关于消息持久化的注意事项​

将消息标记为持久并不能完全保证消息不会丢失。尽管它告诉 RabbitMQ 将消息保存到磁盘，但在 RabbitMQ 接受消息但尚未保存的短时间窗口内仍然存在风险。此外，RabbitMQ 不会为每条消息执行 fsync(2) ——它可能只保存在缓存中而未真正写入磁盘。持久性保证不强，但对于我们简单的任务队列来说已绰绰有余。如果您需要更强的保证，可以使用发布者确认。

公平分发 vs. 轮询分发​

默认情况下，RabbitMQ将按顺序将每条消息发送到下一个消费者。平均而言，每个消费者将获得相同数量的消息。这种分发消息的方式称为轮询。在这种模式下，分发不一定能完全按照我们想要的方式工作。例如，在有两个工作者的情况下，当所有奇数消息都很重而偶数消息都很轻时，一个工作者将一直很忙，而另一个工作者几乎不做任何工作。然而，RabbitMQ对此一无所知，仍然会平均分发消息。

这是因为 RabbitMQ 在消息进入队列时就会分发消息。它不会查看消费者未确认消息的数量。它只是盲目地将第 n 条消息分发给第 n 个消费者。

然而，“公平分发”是Spring AMQP的默认配置。AbstractMessageListenerContainer将DEFAULT_PREFETCH_COUNT的值定义为250。如果将DEFAULT_PREFETCH_COUNT设置为1，其行为将如上所述的轮询传递。

关于prefetchCount = 1的注意事项​

在大多数情况下，prefetchCount等于1过于保守，会严重限制消费者的吞吐量。可以在Spring AMQP 消费者文档中找到适用于此配置的几个例子。

有关预取（prefetch）的更多详细信息，请参阅消费者确认指南。

然而，默认情况下，prefetchCount设置为250，这意味着RabbitMQ一次不会给一个工作者超过250条消息。或者说，当未确认的消息数量为250时，不要将新消息分发给一个工作者。相反，它会将其分发给下一个尚未忙碌的工作者。

期望的prefetchCount值可以通过AbstractMessageListenerContainer.setPrefetchCount(int prefetchCount)设置。

关于队列大小的说明​

如果所有工作进程都忙碌，您的队列可能会填满。您需要密切关注这一点，并可能添加更多工作进程，或采取其他策略。

通过使用Spring AMQP，您可以获得合理的默认配置，用于消息确认和公平分发。Spring AMQP提供的队列的默认持久化和消息的持久化允许消息在RabbitMQ重新启动后仍然得以保留。

有关Channel方法和MessageProperties的更多信息，您可以浏览在线javadocs。要理解Spring AMQP的底层基础，您可以找到rabbitmq-java-client。

现在我们可以继续学习第三个教程，了解如何将同一条消息传递给多个消费者。

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