## 编写一个 WorkerPool 队列消费 > 队列消费是高并发系统中最常用的异步处理模型,通常我们是编写一个 Console 命令行程序在后台执行 Redis、RabbitMQ 等 MQ 的队列消费,并将处理结果落地到 mysql 等数据库中,由于这类需求的标准化比较容易,因此我们开发了 [workerpool](https://github.com/mix-go/workerpool) 库来处理这类需求,基本上大部分异步处理类需求都可使用 首先我们使用 `mix` 命令创建一个 Console 项目骨架: ~~~ mix new --name=hello ~~~ 然后我们在 `commands` 新建 `commands/workerpool.go` 文件: - 定义一个新的命令结构体 `WorkerPoolDaemonCommand` - `WorkerPoolDaemonCommand.Main` 方法为默认执行的入口方法 - 代码中 `workerpool.NewDispatcher(jobQueue, 15, NewWorker)` 创建了一个调度器 - `NewWorker` 负责初始化执行任务的工作协程 - 任务数据会在 `worker.Do` 方法中触发,我们只需要将我们的业务逻辑写到该方法中即可 - 当程序接收到进程信号时,调度器能平滑控制所有的 Worker 在执行完队列里全部的任务后再退出调度,保证数据的完整性 ~~~ package commands import ( "context" "fmt" "github.com/mix-go/console-skeleton/globals" "github.com/mix-go/console/catch" "github.com/mix-go/workerpool" "os" "os/signal" "strings" "syscall" "time" ) type worker struct { workerpool.WorkerTrait } func (t *worker) Do(data interface{}) { defer func() { if err := recover(); err != nil { catch.Error(err) } }() // 执行业务处理 // ... // 将处理结果落地到数据库 // ... } func NewWorker() workerpool.Worker { return &worker{} } type WorkerPoolDaemonCommand struct { } func (t *WorkerPoolDaemonCommand) Main() { redis := globals.Redis() jobQueue := make(chan interface{}, 50) d := workerpool.NewDispatcher(jobQueue, 15, NewWorker) ch := make(chan os.Signal) signal.Notify(ch, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM) go func() { <-ch d.Stop() }() go func() { for { res, err := redis.BRPop(context.Background(), 3*time.Second, "foo").Result() if err != nil { if strings.Contains(err.Error(), "redis: nil") { continue } fmt.Println(fmt.Sprintf("Redis Error: %s", err)) d.Stop(); return } // brPop命令最后一个键才是值 jobQueue <- res[1] } }() d.Run() // 阻塞代码,直到任务全部执行完成并且全部 Worker 停止 } ~~~ 然后我们把上面的 `WorkerPoolDaemonCommand` 在 `manifest/commands` 中注册,新增一个命令配置: - 新建 `manifest/commands/workerpool.go` 文件 ~~~ package commands import ( "github.com/mix-go/console" "github.com/mix-go/console-skeleton/commands" ) func init() { Commands = append(Commands, console.CommandDefinition{ Name: "wpd", Usage: "\tWorker pool daemon demo", Options: []console.OptionDefinition{ { Names: []string{"d", "daemon"}, Usage: "Run in the background", }, }, Command: &commands.WorkerPoolDaemonCommand{}, }, ) } ~~~ ## 编译与测试 > 也可以在 Goland Run 里配置 Program arguments 直接编译执行,[Goland 使用] 章节有详细介绍 接下来我们编译上面的程序: ~~~ // linux & macOS go build -o bin/go_build_main_go main.go // win go build -o bin/go_build_main_go.exe main.go ~~~ 执行 `wpd` 命令: ~~~ $ cd bin $ ./go_build_main_go wpd ~~~ 当我们想在服务器后台执行时,只需增加 `-d/--daemon` 参数: ~~~ $ ./go_build_main_go wpd -d ~~~