RabbitMQ 教程 - 工作队列

工作队列

（使用 Go RabbitMQ 客户端）

信息

前提条件

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

在哪里获得帮助

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

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

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

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

准备工作

在本教程的前一部分，我们发送了一条包含 “Hello World!” 的消息。现在我们将发送代表复杂任务的字符串。我们没有真实的world任务，例如要调整大小的图像或要渲染的 pdf 文件，因此让我们通过假装我们很忙来伪造它 - 通过使用 time.Sleep 函数。我们将字符串中点的数量作为其复杂性；每个点将代表一秒钟的“工作”。例如，用 Hello... 描述的假任务将需要三秒钟。

我们将稍微修改之前示例中的 send.go 代码，以允许从命令行发送任意消息。此程序将计划任务到我们的工作队列，因此我们将其命名为 new_task.go

ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
defer cancel()

body := bodyFrom(os.Args)
err = ch.PublishWithContext(ctx,
  "",           // exchange
  q.Name,       // routing key
  false,        // mandatory
  false,
  amqp.Publishing {
    DeliveryMode: amqp.Persistent,
    ContentType:  "text/plain",
    Body:         []byte(body),
  })
failOnError(err, "Failed to publish a message")
log.Printf(" [x] Sent %s", body)

这是 bodyFrom 函数

func bodyFrom(args []string) string {
	var s string
	if (len(args) < 2) || os.Args[1] == "" {
		s = "hello"
	} else {
		s = strings.Join(args[1:], " ")
	}
	return s
}

我们旧的 receive.go 脚本也需要进行一些更改：它需要为消息正文中的每个点伪造一秒钟的工作。它将从队列中弹出消息并执行任务，因此我们将其称为 worker.go

msgs, err := ch.Consume(
  q.Name, // queue
  "",     // consumer
  true,   // auto-ack
  false,  // exclusive
  false,  // no-local
  false,  // no-wait
  nil,    // args
)
failOnError(err, "Failed to register a consumer")

var forever chan struct{}

go func() {
  for d := range msgs {
    log.Printf("Received a message: %s", d.Body)
    dotCount := bytes.Count(d.Body, []byte("."))
    t := time.Duration(dotCount)
    time.Sleep(t * time.Second)
    log.Printf("Done")
  }
}()

log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever

请注意，我们的假任务模拟了执行时间。

像在第一个教程中一样运行它们

# shell 1
go run worker.go

# shell 2
go run new_task.go

轮询分发

使用任务队列的优势之一是能够轻松地并行化工作。如果我们正在积累工作积压，我们可以只添加更多的工作进程，这样就可以轻松扩展。

首先，让我们尝试同时运行两个 worker.go 脚本。它们都将从队列中获取消息，但具体如何？让我们看看。

您需要打开三个控制台。两个将运行 worker.go 脚本。这些控制台将是我们的两个消费者 - C1 和 C2。

# shell 1
go run worker.go
# => [*] Waiting for messages. To exit press CTRL+C

# shell 2
go run worker.go
# => [*] Waiting for messages. To exit press CTRL+C

在第三个控制台中，我们将发布新任务。启动消费者后，您可以发布一些消息

# shell 3
go run new_task.go First message.
go run new_task.go Second message..
go run new_task.go Third message...
go run new_task.go Fourth message....
go run new_task.go Fifth message.....

让我们看看交付给我们的工作进程的内容

# shell 1
go run worker.go
# => [*] Waiting for messages. To exit press CTRL+C
# => [x] Received 'First message.'
# => [x] Received 'Third message...'
# => [x] Received 'Fifth message.....'

# shell 2
go run worker.go
# => [*] Waiting for messages. To exit press CTRL+C
# => [x] Received 'Second message..'
# => [x] Received 'Fourth message....'

默认情况下，RabbitMQ 会按顺序将每条消息发送给下一个消费者。平均而言，每个消费者将获得相同数量的消息。这种分发消息的方式称为轮询。用三个或更多工作进程尝试一下。

消息确认

执行任务可能需要几秒钟，您可能想知道如果消费者启动一个长时间的任务并在完成之前终止会发生什么。使用我们当前的代码，一旦 RabbitMQ 将消息传递给消费者，它会立即将其标记为删除。在这种情况下，如果您终止一个工作进程，它正在处理的消息将丢失。已分发给此特定工作进程但尚未处理的消息也将丢失。

但是我们不想丢失任何任务。如果工作进程死掉，我们希望将任务传递给另一个工作进程。

为了确保消息永远不会丢失，RabbitMQ 支持消息确认。ack（确认）由消费者发回，以告知 RabbitMQ 已收到并处理了特定消息，并且 RabbitMQ 可以自由删除它。

如果消费者在没有发送 ack 的情况下死掉（其通道关闭、连接关闭或 TCP 连接丢失），RabbitMQ 将理解消息未被完全处理并将重新排队。如果同时有其他消费者在线，它将快速将其重新传递给另一个消费者。这样，即使工作进程偶尔死掉，您也可以确保没有消息丢失。

消费者交付确认会强制执行超时（默认为 30 分钟）。这有助于检测从不确认交付的有缺陷（卡住）的消费者。您可以按照交付确认超时中所述增加此超时。

在本教程中，我们将使用手动消息确认，方法是为 “auto-ack” 参数传递 false，然后在完成任务后，使用 d.Ack(false)（这确认单个交付）从工作进程发送正确的确认。

msgs, err := ch.Consume(
  q.Name, // queue
  "",     // consumer
  false,  // auto-ack
  false,  // exclusive
  false,  // no-local
  false,  // no-wait
  nil,    // args
)
failOnError(err, "Failed to register a consumer")

var forever chan struct{}

go func() {
  for d := range msgs {
    log.Printf("Received a message: %s", d.Body)
    dotCount := bytes.Count(d.Body, []byte("."))
    t := time.Duration(dotCount)
    time.Sleep(t * time.Second)
    log.Printf("Done")
    d.Ack(false)
  }
}()

log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
<-forever

使用此代码，您可以确保即使您在使用 CTRL+C 终止正在处理消息的工作进程时，也不会丢失任何内容。在工作进程终止后不久，所有未确认的消息都会被重新传递。

确认必须在接收交付的同一通道上发送。尝试使用不同的通道进行确认将导致通道级协议异常。请参阅关于确认的文档指南以了解更多信息。

忘记确认

错过 ack 是一个常见的错误。这是一个容易犯的错误，但后果很严重。当您的客户端退出时（这可能看起来像是随机重新交付），消息将被重新交付，但 RabbitMQ 将消耗越来越多的内存，因为它无法释放任何未确认的消息。

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

sudo rabbitmqctl list_queues name messages_ready messages_unacknowledged

在 Windows 上，删除 sudo

rabbitmqctl.bat list_queues name messages_ready messages_unacknowledged

消息持久性

我们已经学习了如何确保即使消费者死掉，任务也不会丢失。但是，如果 RabbitMQ 服务器停止，我们的任务仍然会丢失。

当 RabbitMQ 退出或崩溃时，它会忘记队列和消息，除非您告诉它不要忘记。要确保消息不丢失，需要做两件事：我们需要将队列和消息都标记为持久性。

首先，我们需要确保队列在 RabbitMQ 节点重启后仍然存在。为了做到这一点，我们需要将其声明为持久

q, err := ch.QueueDeclare(
  "hello",      // name
  true,         // durable
  false,        // delete when unused
  false,        // exclusive
  false,        // no-wait
  nil,          // arguments
)
failOnError(err, "Failed to declare a queue")

虽然此命令本身是正确的，但它在我们当前的设置中不起作用。那是因为我们已经定义了一个名为 hello 的队列，它不是持久的。RabbitMQ 不允许您使用不同的参数重新定义现有队列，并且会向尝试这样做的任何程序返回错误。但是有一个快速的解决方法 - 让我们声明一个具有不同名称的队列，例如 task_queue

q, err := ch.QueueDeclare(
  "task_queue", // name
  true,         // durable
  false,        // delete when unused
  false,        // exclusive
  false,        // no-wait
  nil,          // arguments
)
failOnError(err, "Failed to declare a queue")

此 durable 选项更改需要应用于生产者和消费者代码。

此时，我们确信即使 RabbitMQ 重新启动，task_queue 队列也不会丢失。现在我们需要将我们的消息标记为持久性

• 通过使用 amqp.Publishing 采用的 amqp.Persistent 选项。

err = ch.PublishWithContext(ctx,
  "",           // exchange
  q.Name,       // routing key
  false,        // mandatory
  false,
  amqp.Publishing {
    DeliveryMode: amqp.Persistent,
    ContentType:  "text/plain",
    Body:         []byte(body),
})

关于消息持久性的说明

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

公平分发

您可能已经注意到，分发仍然没有完全按照我们想要的方式工作。例如，在有两个工作进程的情况下，当所有奇数消息都很重而偶数消息很轻时，一个工作进程将一直很忙，而另一个工作进程几乎不做任何工作。好吧，RabbitMQ 对此一无所知，并且仍然会均匀地分发消息。

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

为了克服这个问题，我们可以将预取计数设置为值 1。这告诉 RabbitMQ 一次不要给一个工作进程超过一条消息。或者，换句话说，在工作进程处理并确认上一条消息之前，不要向其分发新消息。相反，它会将其分发给下一个不忙的工作进程。

err = ch.Qos(
  1,     // prefetch count
  0,     // prefetch size
  false, // global
)
failOnError(err, "Failed to set QoS")

关于队列大小的说明

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

总结

我们的 new_task.go 类的最终代码

package main

import (
        "context"
        "log"
        "os"
        "strings"
        "time"

        amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
        if err != nil {
                log.Panicf("%s: %s", msg, err)
        }
}

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")
        defer conn.Close()

        ch, err := conn.Channel()
        failOnError(err, "Failed to open a channel")
        defer ch.Close()

        q, err := ch.QueueDeclare(
                "task_queue", // name
                true,         // durable
                false,        // delete when unused
                false,        // exclusive
                false,        // no-wait
                nil,          // arguments
        )
        failOnError(err, "Failed to declare a queue")

        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()

        body := bodyFrom(os.Args)
        err = ch.PublishWithContext(ctx,
                "",           // exchange
                q.Name,       // routing key
                false,        // mandatory
                false,
                amqp.Publishing{
                        DeliveryMode: amqp.Persistent,
                        ContentType:  "text/plain",
                        Body:         []byte(body),
                })
        failOnError(err, "Failed to publish a message")
        log.Printf(" [x] Sent %s", body)
}

func bodyFrom(args []string) string {
        var s string
        if (len(args) < 2) || os.Args[1] == "" {
                s = "hello"
        } else {
                s = strings.Join(args[1:], " ")
        }
        return s
}

（new_task.go 源代码）

以及我们的 worker.go

package main

import (
        "bytes"
        "log"
        "time"

        amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
        if err != nil {
                log.Panicf("%s: %s", msg, err)
        }
}

func main() {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        failOnError(err, "Failed to connect to RabbitMQ")
        defer conn.Close()

        ch, err := conn.Channel()
        failOnError(err, "Failed to open a channel")
        defer ch.Close()

        q, err := ch.QueueDeclare(
                "task_queue", // name
                true,         // durable
                false,        // delete when unused
                false,        // exclusive
                false,        // no-wait
                nil,          // arguments
        )
        failOnError(err, "Failed to declare a queue")

        err = ch.Qos(
                1,     // prefetch count
                0,     // prefetch size
                false, // global
        )
        failOnError(err, "Failed to set QoS")

        msgs, err := ch.Consume(
                q.Name, // queue
                "",     // consumer
                false,  // auto-ack
                false,  // exclusive
                false,  // no-local
                false,  // no-wait
                nil,    // args
        )
        failOnError(err, "Failed to register a consumer")

        var forever chan struct{}

        go func() {
                for d := range msgs {
                        log.Printf("Received a message: %s", d.Body)
                        dotCount := bytes.Count(d.Body, []byte("."))
                        t := time.Duration(dotCount)
                        time.Sleep(t * time.Second)
                        log.Printf("Done")
                        d.Ack(false)
                }
        }()

        log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
        <-forever
}

（worker.go 源代码）

使用消息确认和预取计数，您可以设置工作队列。即使 RabbitMQ 重新启动，持久性选项也使任务能够继续存在。

有关 amqp.Channel 方法和消息属性的更多信息，您可以浏览 amqp API 参考。

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