- xkafkaは、Go環境でKafkaをHTTPサービスのようにシンプルに使えるようにするオープンソースライブラリ
- 従来のconfluent-kafka-goでは複雑な処理ループと大量のボイラープレートコードが必要だったが、xkafkaはHandler、Middleware、Message構造によってコアロジックに集中できるようにする
- メッセージの発行と消費をHTTPのリクエスト/レスポンスのように直感的に扱え、オフセット管理、並行性設定、エラーハンドリングなどKafkaの複雑さを大きく隠蔽する
- Streaming/Batch処理、逐次/非同期処理、At-most-once/At-least-once保証など、本番サービスで求められるさまざまなパターンを簡単にサポートする
- 階層的なエラー処理、ミドルウェアベースのリトライ/ロギング/メトリクスなど、実務で必要なパターンを容易に適用できる
HTTP-like Kafka
- xkafkaは、GoでKafkaをHTTPサービスのように抽象化するライブラリ
- MessageはHTTPリクエストに似ており、トピック/パーティション/オフセット/キー/値/ヘッダー/コールバックなどを含む
- HandlerはHTTP Handlerのようにビジネスロジックを処理する
- Middlewareはロギング、メトリクス、リトライなどの付加機能をビジネスロジックから分離して適用できる
メッセージ発行 (Publishing Messages)
xkafka.NewProducerでProducerを作成し、メッセージオブジェクトを作ってPublish関数で発行する- 非同期発行（AsyncPublish）やコールバック登録が可能で、高性能処理や非同期イベント処理を容易に実装できる
- バックグラウンドgoroutineでメッセージ配送を処理し、コールバックで配送状態を追跡できる
メッセージ消費 (Consuming Messages)
- Consumer作成時に、Handler関数とトピック/ブローカー/設定などを指定する
consumer.Use()でミドルウェアを追加できるconsumer.Run(ctx)でメッセージ消費を開始する
Streaming vs. Batch
- Streaming: メッセージ到着ごとに即座に1件ずつ処理。スループットが低い場合やメモリ節約、強い処理保証に有利
- Batch: 一定件数または一定時間単位でまとめて処理。高スループットシステムや下流負荷の緩和に有利
Sequential or Async
- デフォルトは逐次処理（Sequential）— 1件の処理が終わってから次のメッセージを読む
xkafka.Concurrency(N)を使うと、N件のメッセージ（またはバッチ）を同時処理する非同期（Async）モードをサポート
オフセット管理
- Kafkaの標準動作では、メッセージ配信直後にオフセットが先へ進むため、障害時にメッセージ損失の可能性がある
- xkafkaは
enable.auto.offset.store=falseを設定し、メッセージ（またはバッチ）の処理完了後にのみオフセットを保存する
- 別途DBやキューでメッセージ状態を管理しなくても、Kafka上で処理保証を実現できる
-
At-Most-Once Guarantee
- 基本的にはKafkaの
enable.auto.commit=trueに従い、バックグラウンドでオフセットをコミットする
xkafka.ManualCommit(true)と逐次処理により、各メッセージ/バッチを読む前にオフセットをコミットしてAt-most-onceを保証する
-
At-Least-Once Guarantee
xkafka.ManualCommit(true)と並行性（N>1）を組み合わせることで、並列処理中でもオフセットを同期的かつ順序通りにコミットする- At-least-once保証パターンを簡単に適用できる
エラーハンドリング
-
Handlerレベル
- Handler内でアプリケーションエラーの処理やDead Letter Queueへの送信などを行える
- 成功時は
msg.AckSuccess()、スキップ時はmsg.AckSkip()、失敗時はmsg.AckFail(err)のように明示的に制御する
-
Middlewareレベル
- ミドルウェアでリトライ、エラーロギングなどの共通ロジックを複数のHandlerに再利用できる
- さまざまなエラーに応じて異なるリトライ方針や処理方法を容易に適用できる
-
Globalレベル
- Kafkaブローカー/ライブラリエラーは必須オプションの
xkafka.ErrorHandlerで一元処理する
- このハンドラーがnon-nilエラーを返した場合、Consumer/Producerの動作を停止する
結論
- xkafkaは、Apache Kafkaの複雑な利用体験をGo開発者になじみのあるHTTPサーバー構造へと変える
- 不要なボイラープレートを減らし、ビジネスロジックだけに集中できる環境を提供する
- 既存のconfluent-kafka-goコードと比べて、はるかに簡潔で直感的
- 公式ドキュメントとサンプルを参照してすぐに始められる
1件のコメント
うーん、golang では
saramaのほうがより好まれていると知っていたのですが……思った以上に Kafka クライアントは……ブローカー障害や例外時には非常に複雑なので、
あらゆるケースをカバーできるのか……