goworker - Golang高效后台任务处理插件库

goworker是一个兼容Resque的、基于Go的后台工作程序。它允许您使用Ruby等表达性语言将作业推入队列，同时利用Go的效率和并发性来最小化作业延迟和成本。

安装

要安装goworker，使用以下命令：

go get github.com/benmanns/goworker

然后在您的worker中导入：

import "github.com/benmanns/goworker"

快速入门

基本使用

要创建一个worker，编写一个匹配以下签名的函数：

func(string, ...interface{}) error

并使用Register方法注册它：

goworker.Register("MyClass", myFunc)

下面是一个打印参数的简单worker示例：

package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func myFunc(queue string, args ...interface{}) error {
	fmt.Printf("From %s, %v\n", queue, args)
	return nil
}

func init() {
	goworker.Register("MyClass", myFunc)
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}

共享资源的Worker

要创建共享数据库池或其他资源的worker，可以使用闭包来共享变量：

package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func newMyFunc(uri string) (func(queue string, args ...interface{}) error) {
	foo := NewFoo(uri)
	return func(queue string, args ...interface{}) error {
		foo.Bar(args)
		return nil
	}
}

func init() {
	goworker.Register("MyClass", newMyFunc("http://www.example.com/"))
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}

带配置的Worker

下面是一个带配置的worker示例：

package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func myFunc(queue string, args ...interface{}) error {
	fmt.Printf("From %s, %v\n", queue, args)
	return nil
}

func init() {
	settings := goworker.WorkerSettings{
		URI:            "redis://localhost:6379/",
		Connections:    100,
		Queues:         []string{"myqueue", "delimited", "queues"},
		UseNumber:      true,
		ExitOnComplete: false,
		Concurrency:    2,
		Namespace:      "resque:",
		Interval:       5.0,
	}
	goworker.SetSettings(settings)
	goworker.Register("MyClass", myFunc)
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}

参数处理

goworker工作函数接收它们服务的队列和一个接口切片。要将它们用作其他函数的参数，请使用Go类型断言将它们转换为可用的类型：

// 期望 (int, string, float64)
func myFunc(queue, args ...interface{}) error {
	idNum, ok := args[0].(json.Number)
	if !ok {
		return errorInvalidParam
	}
	id, err := idNum.Int64()
	if err != nil {
		return errorInvalidParam
	}
	name, ok := args[1].(string)
	if !ok {
		return errorInvalidParam
	}
	weightNum, ok := args[2].(json.Number)
	if !ok {
		return errorInvalidParam
	}
	weight, err := weightNum.Float64()
	if err != nil {
		return errorInvalidParam
	}
	doSomething(id, name, weight)
	return nil
}

测试

对于测试，使用redis-cli程序将作业插入Redis队列很有帮助：

redis-cli -r 100 RPUSH resque:queue:myqueue '{"class":"MyClass","args":["hi","there"]}'

这将在myqueue队列中插入100个MyClass worker的作业。它等同于：

class MyClass
  @queue = :myqueue
end

100.times do
  Resque.enqueue MyClass, ['hi', 'there']
end

或者使用Go代码：

goworker.Enqueue(&goworker.Job{
    Queue: "myqueue",
    Payload: goworker.Payload{
        Class: "MyClass",
        Args: []interface{}{"hi", "there"},
    },
})

配置选项

有几个标志控制goworker客户端的操作：

• -queues="comma,delimited,queues" - 唯一必需的标志。建议做法是使用不同的队列将Resque worker与goworker分开。
• -interval=5.0 - 指定在没有作业时轮询之间的等待时间。
• -concurrency=25 - 指定并发执行worker的数量。
• -connections=2 - 指定goworker将在轮询器和所有worker之间消耗的最大Redis连接数。
• -uri=redis://localhost:6379/ - 指定goworker从中轮询作业的Redis数据库的URI。
• -namespace=resque: - 指定goworker从中检索作业并存储worker统计信息的命名空间。
• -exit-on-complete=false - 当队列中没有作业剩余时退出goworker。

信号处理

要停止goworker，向进程发送QUIT、TERM或INT信号。这将立即停止作业轮询。当前最多可以有$CONCURRENCY个作业正在运行，这些作业将继续运行直到完成。

故障模式

与Resque一样，goworker不保证在进程关闭时作业的安全性。worker必须既是幂等的，又能在失败时容忍作业丢失。

如果进程被KILL或系统故障杀死，轮询器缓冲区中可能有一个当前作业将丢失，队列或worker变量中没有任何表示。