Golang Go语言中hibiken消息队列如何控制queue的并发

https://github.com/hibiken/asynq

看文档，

	srv := asynq.NewServer(
		asynq.RedisClientOpt{Addr: ":6379"},
		asynq.Config{Concurrency: 20}, //并发控制
	)

h := asynq.NewServeMux()
// ... Register handlers

// Run blocks and waits for os signal to terminate the program.
if err := srv.Run(h); err != nil {
	log.Fatal(err)
}

这个是对于整个项目的并发，现在只需要对于不同任务的并发控制，不知道怎么实现。


Golang Go语言中hibiken消息队列如何控制queue的并发

更多关于Golang Go语言中hibiken消息队列如何控制queue的并发的实战教程也可以访问 https://www.itying.com/category-94-b0.html

6 回复

可以控制队列的优先级，也就变相的控制了某个队列的并发数：

srv := asynq.NewServer(
asynq.RedisClientOpt{Addr: redisAddr},
asynq.Config{
// Specify how many concurrent workers to use
Concurrency: 10,
// Optionally specify multiple queues with different priority.
Queues: map[string]int{
“queue1”: 6,
“queue2”: 3,
“queue3”: 1,
},
// See the godoc for other configuration options
},
)

更多关于Golang Go语言中hibiken消息队列如何控制queue的并发的实战系列教程也可以访问 https://www.itying.com/category-94-b0.html


https://github.com/hibiken/asynq/wiki/Queue-Priority#strict-priority

这种勉强能实现，但是不太正确。

找到解决方案了，可以再重新起一个 server

然后记得指定就可以了
client.Enqueue(task1, asynq.Queue(“notifications”))

这个 server 专门跑 notifications 队列？

不是，是一些牵扯到扣费的逻辑，为了保证用户余额的一致性，然后串行扣费。

在Golang中，使用hibiken消息队列库（通常指的是go-nats，这是一个流行的NATS客户端库）时，控制队列的并发通常涉及对并发访问的适当管理和消息处理的并行化。以下是一些关键策略：

  1. 使用Worker Pool模式：创建一个固定大小的goroutine池来处理消息。这可以通过channel和goroutine的结合来实现，确保消息处理在控制的并发级别下进行。

  2. 消息处理函数内部同步：如果消息处理逻辑本身不是线程安全的，确保在函数内部使用适当的同步机制（如mutexes或channels）来保护共享资源。

  3. 调整NATS订阅选项：NATS客户端允许设置一些订阅选项，比如队列组（Queue Group），它可以在多个客户端之间分配消息，确保每个消息只被处理一次。此外，可以调整最大订阅连接数等参数来控制并发。

  4. 利用Go的并发特性：Go的goroutines和channels是处理并发消息队列的自然选择。通过启动多个goroutines来监听和处理消息，可以充分利用多核CPU，同时保持代码的简洁和可读性。

  5. 监控和调优：在实际部署后，通过监控系统的性能指标（如CPU使用率、内存占用、消息处理延迟等）来调整并发级别和消息处理逻辑，以达到最佳性能。

综上所述，控制hibiken消息队列（或NATS客户端）的并发需要综合考虑并发模型、同步机制、NATS配置以及Go语言的并发特性。

回到顶部