我正在尝试找到一种从输入队列异步使用、使用多个工作程序处理内容然后发布到输出队列的好方法。到目前为止，我已经尝试了很多示例，最近使用的代码来自 here和 here作为灵感。

然而，我当前的代码似乎没有做它应该做的事情，增加工作人员的数量不会提高性能(消耗或发布的 msg/s)，并且 goroutines 的数量在运行时保持相当稳定。

主要内容:

func main() {
    maxWorkers := 10

    // channel for jobs
    in := make(chan []byte)
    out := make(chan []byte)

    // start workers
    wg := &sync.WaitGroup{}
    wg.Add(maxWorkers)
    for i := 1; i <= maxWorkers; i++ {
        log.Println(i)
        defer wg.Done()
        go processor(in, out)
    }

    // add jobs
    go collector(in)
    go sender(out)

    // wait for workers to complete
    wg.Wait()
}

收集器基本上是来自 RabbitMQ 站点的示例，带有一个 goroutine，它从队列中收集消息并将它们放在“in” channel 上:

forever := make(chan bool)
go func() {
    for d := range msgs {
        in <- d.Body
        d.Ack(false)
    }
}()
log.Printf("[*] Waiting for messages. To exit press CTRL+C")
<-forever

处理器接收“输入”和“输出” channel ，解码 JSON，执行一系列正则表达式，然后将输出放入“输出” channel :

func processor(in chan []byte, out chan []byte) {

    var (
    // list of regexes declared here
    )

    for {
        body := <-in

        jsonIn := &Data{}
        err := json.Unmarshal(body, jsonIn)
        if err != nil {
            log.Fatalln("Failed to decode:", err)
        }

        content := jsonIn.Content

        //process regexes using:
        //jsonIn.a = r1.FindAllString(content, -1)

        jsonOut, _ := json.Marshal(jsonIn)

        out <- jsonOut
    }
}

最后，发送方只是来自 RabbitMQ 站点的代码，设置连接，从“输出” channel 读取数据，然后发布到 RMQ 队列:

for {
    jsonOut := <-out

    err = ch.Publish(
        "",     // exchange
        q.Name, // routing key
        false,  // mandatory
        false,
        amqp.Publishing{
            DeliveryMode: amqp.Persistent,
            ContentType:  "text/json",
            Body:         []byte(jsonOut),
        })
    failOnError(err, "Failed to publish a message")

}

这是一个我会经常使用的模式，所以我花了很多时间试图找到正确(并且很好)工作的东西 - 任何建议或帮助将不胜感激(如果它并不明显，我是 Go 的新手)。

最佳答案

有几件事跳出来了:

在主函数中完成

wg.Add(maxWorkers)
for i := 1; i <= maxWorkers; i++ {
    log.Println(i)
    defer wg.Done()
    go processor(in, out)
}

此处的 defer 是在 main 返回时执行的，因此它实际上并不表示处理何时完成。不过，我认为这不会对您的程序的性能概况产生影响。

要解决这个问题，您可以将 wg *sync.WaitGroup 传递给您的处理器，以便您的处理器可以指示它何时完成。

CPU 绑定(bind)处理

解析消息和执行 Regex 是 CPU 密集型工作负载。你的机器是几核的？如果您在两台不同的机器上运行您的程序，吞吐量会受到怎样的影响，吞吐量是 2 倍吗？如果将内核数量加倍怎么办？用 1 个 worker 和 2 个处理器 worker 运行你的程序怎么样？这会使吞吐量翻倍吗？你的 rabbitmq 本地实例用完了吗？是瓶颈吗？？

设置基准测试和负载测试工具应该允许您设置实验以查看瓶颈所在:)

对于基于队列的服务，很容易设置一个测试工具来为 rabbitmq 填充一组积压和基准测试你可以多快地处理这些消息，或者设置一个负载生成器以每秒发送 x 条消息到 rabbitmq 并观察你是否可以跟上。

rabbitmq 对消息处理吞吐量有很好的可见性吗？如果不是，我经常添加一个计数器来执行代码，然后记录一个时间间隔的总体平均吞吐量，以大致了解性能:

start := time.Now()
updateInterval := time.Tick(1 * time.Second)
numIn := 0
for {
    select {
    case <-updateInterval:
        log.Infof("IN - Count: %d", numIn)
        log.Infof("IN - Througput: %.0f events/second",
            float64(numIn)/(time.Now().Sub(start)).Seconds())
    case e := <-msgs:
        numIn++
        in <- d.Body
        d.Ack(false)
    }
}

关于go - RabbitMQ 多 worker 模式，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/46448741/