Rive: Go向け高速で信頼性の高いバックグラウンドジョブシステム

River Riverは、GoとPostgres向けに堅牢で高性能なジョブ処理システムです。ホームページ、ドキュメント、godoc、およびRiver UIとそのライブデモをご覧ください。 Postgres用に構築されているRiverは、アプリケーションデータとジョブキューに同じデータベースを使用することを推奨しています。他のデータベース変更とトランザクション的にジョブをエンキューすることで、分散システムのあらゆる問題が回避されます。ジョブは、トランザクションがコミットされた場合にエンキューされることが保証され、トランザクションがロールバックされた場合は削除され、コミットされるまで作業に表示されません。この哲学の詳細については、トランザクションエンキューを参照してください。 ジョブの引数とワーカー ジョブは構造体ペアで定義され、JobArgsの実装とWorkerの実装を持ちます。ジョブ引数はJSONアノテーションを含み、ジョブがデータベースにシリアル化される方法と、ジョブを一意に識別する安定した文字列である「種類(kind)」を定義します。 ```go type SortArgs struct { // Stringsはソートする文字列のスライスです。 Strings []string `json:"strings"` } func (SortArgs) Kind() string { return "sort" } ``` ワーカーは、ジョブがどのように実行されるかを規定するWork関数を公開します。 ```go type SortWorker struct { // 埋め込みWorkerDefaultsは、Workerインターフェースの残りのメソッドを // 満たすためのデフォルトメソッドを設定します。 river.WorkerDefaults[SortArgs] } func (w *SortWorker) Work(ctx context.Context, job *river.Job[SortArgs]) error { sort.Strings(job.Args.Strings) fmt.Printf("Sorted strings: %+v\n", job.Args.Strings) return nil } ``` ワーカーの登録 ジョブは「種類(kind)」文字列によって一意に識別されます。ワーカーは起動時に登録され、Riverがジョブをワーカーに割り当てる方法を認識できるようにします。 ```go workers := river.NewWorkers() // ワーカーがすでに登録されているか、無効な場合はAddWorkerはパニックします。 river.AddWorker(workers, &SortWorker{}) ``` クライアントの開始 River Clientはジョブ挿入のためのインターフェースを提供し、ジョブ処理およびメンテナンスサービスを管理します。クライアントは、データベースプール、ドライバー、およびWorkersバンドルとその他の設定を含むconfig構造体で作成されます。以下は、1つのキュー（「default」）で一度に最大100のワーカーゴルーチンを処理するクライアントの例です。 ```go riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{ Queues: map[string]river.QueueConfig{ river.QueueDefault: {MaxWorkers: 100}, }, Workers: workers, }) if err != nil { panic(err) } // クライアントをインラインで実行します。実行されるすべてのジョブはctxを継承します。 if err := riverClient.Start(ctx); err != nil { panic(err) } ``` ワーカーは省略することもできますが、挿入されたジョブの種類を実行できるワーカーがあることをRiverがチェックできるように、含めるのがより良い方法です。 停止 プログラムのシャットダウン時にもクライアントを停止する必要があります。これにはいくつかの方法がありますが（グレースフルシャットダウンを参照）、最も短い方法は、プログラムが停止する準備ができたときに`Start`に送られたコンテキストをキャンセルすることです。例えば、SIGINT/SIGTERMで停止する場合： ```go riverClient, err := river.NewClient(riverpgxv5.New(dbPool), &river.Config{ SoftStopTimeout: 10 * time.Second, ... }) if err != nil { panic(err) } signalCtx, stop := signal.NotifyContext(ctx, syscall.SIGINT, syscall.SIGTERM) defer stop() // 新しい作業のフェッチを停止し、アクティブなジョブが終了するのを待ちます。 // SoftStopTimeoutが経過した後、ジョブをキャンセルします。 if err := riverClient.Start(signalCtx); err != nil { panic(err) } <-riverClient.Stopped() ``` あるいは、明示的な`Stop`呼び出しを使用します。 ```go if err := riverClient.Stop(ctx); err != nil { panic(err) } ``` 挿入のみを行うクライアントはジョブを挿入しますが、処理は行わず、起動や停止は必要ありません。 ジョブの挿入 `Client.InsertTx`は、ジョブ引数のインスタンスと組み合わせて、トランザクションで作業するジョブを挿入するために使用されます。 ```go _, err = riverClient.InsertTx(ctx, tx, SortArgs{ Strings: []string{ "whale", "tiger", "bear", }, }, nil) if err != nil { panic(err) } ``` 完全なコードについては、InsertAndWorkの例をご覧ください。 その他の機能 * Postgres `COPY FROM`を使用して多数のジョブを効率的に一括挿入する機能。 * ワーク関数内からジョブをキャンセルする機能。 * エラーとパニック処理。 * 複数のキューにより、ジョブのスループット、ワーカーの可用性、コンポーネント間の分離をより良く保証。 * 定期的なジョブとcronジョブ。 * 将来のスケジュールされた時刻に自動的に実行されるスケジュールされたジョブ。 * ワーク関数内からジョブをスヌーズする機能。 * キューのアクティビティと統計へのサブスクリプション。ロギングやメトリクスなどのテレメトリーのための簡単なフックを提供。 * ジョブが期待通りに挿入されることを確認するためのテストヘルパー。 * トランザクションの他の変更と共にジョブの完了がコミットされることを保証するトランザクションジョブ完了。 * 引数、期間、キュー、状態によるユニークなジョブ。 * ジョブとキューを検査および操作するためのWeb UI。 * ワーカー実装を簡素化するワーク関数。 クロス言語エンキュー Riverは、一部のGo以外の言語でのジョブ挿入をサポートしており、それらはGo実装によって処理されます。これは、Goの高速なランタイムを活用してジョブがパフォーマンスを必要とする場合に望ましいかもしれません。 * Pythonからのジョブ挿入。 * Rubyからのジョブ挿入。 開発 Riverの開発方法をご覧ください。 感謝 Riverは、長年にわたる他のバックグラウンドジョブライブラリでの経験、特に以下のものに大きくインスパイアされました。 * ElixirのOban。 * RubyのQue、Sidekiq、Delayed::Job、GoodJob。 * .NETのHangfire。 ソフトウェアエコシステムを前進させていただき、ありがとうございます。