go-workers を使ってみた

• 概要
• 環境
• エンキューするクライアントの作成 (golang)
• golang のワーカーの作成
• 試してみる
• ruby のワーカーの作成
• go-workers のミドルウェア機能
• 最後に
• 参考サイト

概要

golang の go-workers は Sidekiq と互換があり Ruby のワーカーに enqueue することができます
今回は簡単な go-wokers の使い方と Ruby の sidekiq ワーカーとの連携をやってみたので紹介します

環境

• CentOS 7.5.1804
• redis 3.2.12
• golang 1.11.2
  
  • go-wokers dbf81d0b75bbe2fd90ef66a07643dd70cb42a88a
• Ruby 2.5.0p0
  
  • sidekiq 5.2.5

エンキューするクライアントの作成 (golang)

まずは Redis にジョブをエンキューするクライアントを作成します

• vim main.go

package main

import "github.com/jrallison/go-workers"

func main() {
        workers.Configure(map[string]string{
                "server":   "localhost:6379",
                "process": "1",
        })
        workers.Enqueue("default", "SampleWorker", []int{1, 2})
}

Enqueue メソッドを使います
引数はそれぞれ「キュー名」「ワーカーのクラス名」「ワーカに渡す値」になります
Ruby の sidekiq の場合デフォルトのキュー名が default になっているので、それを指定します

golang のワーカーの作成

ではワーカーを作成します
まずは golang 製のワーカーを作成します
処理としてはエンキューされたジョブの ID と与えられた値を取得して表示するだけです

• vim worker.go

package main

import (
    "fmt"
    "github.com/jrallison/go-workers"
)

func perform(msg *workers.Msg) {
    fmt.Println("perform")
    fmt.Println(msg.Jid())
    a, _ := msg.Args().Array()
    fmt.Println(a[0])
}

func main() {
    workers.Configure(map[string]string{
        "server":  "localhost:6379",
        "process": "1",
    })
    workers.Process("default", perform, 1)
    workers.Run()
}

ワーカーのメイン処理は workers.Process で定義します
1 つ目の引数でキューを指定します
2 つ目の引数でメインの処理をする関数 (perform) を指定します
3 つ目の引数でワーカーの並列数を指定します
workers.Run() でワーカーを起動しプロセスとして常駐させます

試してみる

まずはワーカーを起動します

• go run worker.go

workers: 2019/02/13 08:32:20.330636 processing queue default with 1 workers.

こんな感じで起動すると思います
ワーカーが無事起動したらエンキューするクライアントを実行してみましょう

• go run main.go

するとジョブが default キューにエンキューされワーカー側でジョブが実行されるのが確認できると思います

workers: 2019/02/13 08:33:54.426224 default JID-0467cf107c249376f4ad84f4 start
workers: 2019/02/13 08:33:54.426303 default JID-0467cf107c249376f4ad84f4 args: [1,2]
perform
0467cf107c249376f4ad84f4
1
workers: 2019/02/13 08:33:54.427259 default JID-0467cf107c249376f4ad84f4 done: 1.031331ms

ジョブ ID やジョブの引数の値が表示されれば OK です

ruby のワーカーの作成

次に Sidekiq を使って Ruby 製のワーカーを作成してみたいと思います

• bundle init
• vim Gemfile

gem "sidekiq"

• bundle install --path vendor

ちなみに dep を使っている場合は golang も vendor ディレクトリを作成するため golang のライブラリを上書きしてしまいます
なので ruby ワーカーは別ディレクトリで作業しましょう

• vim worker.rb

require 'sidekiq'

class SampleWorker
  include Sidekiq::Worker

  def perform(*opts)
    puts "perform"
    puts self.jid
    puts opts[0]
  end
end

Ruby だと少し簡潔に書けます
起動しておきましょう

• bundle exec sidekiq -r ./worker.rb

これでエンキューワーカーを起動すると Ruby 製のワーカーがジョブを実行すると思います

go-workers のミドルウェア機能

go-workers にはミドルウェア機能がありこれを使うことでジョブを実行する前後の処理を記載することができます
golang のワーカーを以下のように修正します

package main

import (
    "fmt"
    "github.com/jrallison/go-workers"
)

func perform(msg *workers.Msg) {
    fmt.Println("perform")
    fmt.Println(msg.Jid())
    a, _ := msg.Args().Array()
    fmt.Println(a[0])
}

type myMiddleware struct {
}

func (m *myMiddleware) Call(queue string, message *workers.Msg, next func() bool) (acknowledge bool) {
    fmt.Println("pre commit")
    acknowledge = next()
    fmt.Println("post commit")
    return
}

func main() {
    workers.Configure(map[string]string{
        "server":  "localhost:6379",
        "process": "1",
    })
    workers.Middleware.Append(&myMiddleware{})
    workers.Process("default", perform, 1)
    workers.Run()
}

myMiddleware という構造体を作成しその構造体に Call というメソッドを実装します
そしてワーカーにミドルウェアを追加します (workers.Middleware.Append)
Call メソッド内で必ず記載が必要になるのが acknowledge = next() の部分です
これでワーカーのメイン処理が実行される部分になります
なのでその前後に行いたい前処理を記載することでジョブ共通の前後の処理を実装することができます

今回は構造体に特にフィールドの定義はありませんがフィールドを定義しても OK です

最後に

go-workers を使って golang と ruby 製のワーカーをそれぞれ作成してみました
これを使えば API を golang で記載してワーカーは Ruby で記載するなども可能です
結局 Sidekiq は Redis のBRPOPLPUSH なのでそれを使えればどんな言語でも書けるのですがやはりライブラリがある方が簡単だと思います

参考サイト

• https://github.com/jrallison/go-workers