Golang 言語の Kafka クライアント ライブラリ Sarama

01. はじめに

Apache Kafka はオープンソースのメッセージング エンジン システムです。プロジェクトにおけるその主な役割は、ピーク負荷を軽減し、谷を埋めて分離することです。この記事では、Apache Kafka 用の Golang クライアント ライブラリである Sarama のみを紹介します。 Sarama は、Apache Kafka 0.8 以降用の MIT ライセンスの Golang クライアント ライブラリです。

Apache Kafka サーバーに慣れていない場合は、まず公式ドキュメントの「Getting Started」セクションを読むことをお勧めします。この記事では、Apache Kafka 2.8 バージョンを使用します。

[[397879]]

02. プロデューサー

Sarama ライブラリの AsyncProducer または SyncProducer を使用してメッセージを生成できます。ほとんどの場合、メッセージを生成するには AsyncProducer を使用することをお勧めします。チャネルを通じてメッセージを受信し、バックグラウンドで可能な限り効率的に非同期的にメッセージを生成します。

SyncProducer は、Kafka メッセージを送信した後、ACK 確認を受信するまでブロックします。 SyncProducer には 2 つの注意点があります。一般的に効率が低く、実際の耐久性の保証は Producer.RequiredAcks の構成値によって異なります。構成によっては、SyncProducer によって確認されたメッセージが失われる場合もありますが、使用方法はより簡単になります。

読者の理解を深めるために、この記事では SyncProducer をプロデューサーとして使用する方法を紹介します。 AsyncProducer をプロデューサーとして使用する方法を知りたい読者は、公式ドキュメントを参照してください。

SyncProducer をプロデューサーとして使用するサンプル コード:

 func sendMessage (brokerAddr []string, config *sarama.Config, topic string, value sarama.Encoder) {
 プロデューサー、エラー:= sarama.NewSyncProducer(brokerAddr, config)
 err != nil の場合 {
  fmt.Println(エラー)
戻る 
 }
 遅延関数() {
  err = プロデューサーの場合。近い（）;エラー != ゼロ {
   fmt.Println(エラー)
戻る 
  }
 }()
 メッセージ:= &sarama.ProducerMessage{
  トピック: トピック、
  値: 値、
 }
 パーティション、オフセット、エラー:=producer.SendMessage(msg)
 err != nil の場合 {
  fmt.Println(エラー)
戻る 
 }
 fmt.Printf( "パーティション:%d オフセット:%d\n" , パーティション, オフセット)
 }

上記のコードを読むと、ブローカーのアドレスと構成情報を指定して、NewSyncProducer() を呼び出して新しい SyncProducer を作成します。 SendMessage() を呼び出すと、指定されたメッセージが生成され、生成が成功したか失敗したかの場合にのみ戻ります。生成されたメッセージのパーティションとオフセットを返します。メッセージの生成が失敗した場合はエラーを返します。

リークを回避するには、プロデューサーで Close() を呼び出す必要があることに注意することが重要です。スコープ外になったときに自動的にガベージ コレクションされない可能性があるためです。

03. 消費者

Sarama ライブラリの Consumer または ConsumerGroup API を使用してメッセージを消費できます。読者の理解を容易にするために、この記事では Consumer を使用してメッセージを消費する方法を紹介します。

Consumer は、ブローカーからの Kafka メッセージを処理する PartitionConsumers を管理します。

メッセージを消費するコンシューマーのサンプル コード:

 func consumer (brokenAddr []string, topic string, partition int32, offset int64) {
 消費者、エラー:= sarama.NewConsumer(brokenAddr, nil)
 err != nil の場合 {
  fmt.Println(エラー)
戻る 
 }
 遅延関数() {
  err = コンシューマーの場合。近い（）;エラー != ゼロ {
   fmt.Println(エラー)
戻る 
  }
 }()
 パーティションコンシューマー、エラー:= consumer.ConsumePartition(トピック、パーティション、オフセット)
 err != nil の場合 {
  fmt.Println(エラー)
戻る 
 }
 遅延関数() {
  err = partitionConsumer の場合。近い（）;エラー != ゼロ {
   fmt.Println(エラー)
戻る 
  }
 }()
 msg := range の場合、partitionConsumer.Messages() {
  fmt.Printf( "パーティション:%d オフセット:%d キー:%s 値:%s\n" , msg.Partition, msg.Offset, msg.Key , msg.Value)
 }
 }

上記のコードを読むと、ブローカーのアドレスと構成情報を指定して、NewConsumer() を呼び出して新しいコンシューマーを作成します。トピック、パーティション、オフセットを指定して、ConsumePartition() を呼び出して PartitionConsumer を作成します。 PartitionConsumer は、指定されたトピックとパーティションからの Kafka メッセージを処理します。

リークを防ぐために、consumer とpartitionConsumer で Close() を呼び出す必要があることに注意することが重要です。スコープ外になったときに、自動的にガベージ コレクションされない可能性があるためです。

04. 結論

この記事では主に、Apache Kafka の Golang 言語クライアント ライブラリ Sarama を使用して Kafka メッセージを生成および消費する方法を紹介します。生産者と消費者の両方に対して簡単な例を示します。さらに、Sarama ライブラリは他の多くの API も提供します。興味のある読者は、公式ドキュメントを読んで詳細を確認してください。