Go: Apache Kafka® のサンプルコード¶

このチュートリアルでは、Apache Kafka® クラスターにメッセージを生成し、このクラスターからメッセージを消費する Golang クライアントアプリケーションを実行します。

チュートリアルの実行後に、提供されたソースコードをリファレンスとして使用して、独自の Kafka クライアントアプリケーションを開発します。

前提条件¶

クライアント¶

Apache Kafka 向け Confluent Golang クライアントをインストールした、正常に機能する Go 環境

SSL トラストストアの構成¶

オペレーティングシステムまたは Linux ディストリビューションによっては、追加の手順を実行して SSL CA ルート証明書をセットアップする必要がある場合があります。システムで SSL CA ルート証明書が正しくセットアップされていない場合は、次のような SSL handshake failed エラーメッセージが表示される場合があります。

%3|1605776788.619|FAIL|rdkafka#producer-1| [thrd:sasl_ssl://...confluent.cloud:9092/bootstr]: sasl_ssl://...confluent.cloud:9092/bootstrap: SSL handshake failed: error:14090086:SSL routines:ssl3_get_server_certificate:certificate verify failed: broker certificate could not be verified, verify that ssl.ca.location is correctly configured or root CA certificates are installed (brew install openssl) (after 258ms in state CONNECT)
Copy

この場合は、検証済み CA ルート証明書のセットを手作業でインストールする必要があります。また、クライアントコードを変更して ssl.ca.location 構成プロパティを設定する必要がある場合もあります（詳細については、このクライアントの基盤である librdkafka のドキュメントを参照してください）。

macOS¶

より新しい macOS バージョン（10.15 など）では、依存関係の追加が必要な場合があります。

Python クライアントの場合:

pip install certifi
Copy

その他のクライアントの場合:

brew install openssl
Copy

CA ルート証明書をインストールしたら、クライアントコードで ssl.ca.location プロパティを設定します。プロデューサーとコンシューマーのコードファイルを両方とも編集し、プロデューサーおよびコンシューマーのプロパティ内に ssl.ca.location 構成パラメーターを追加します。その値は、ホスト上の適切な CA ルート証明書ファイルの場所に対応している必要があります。

Python クライアントの場合は、certifi.where() を使用して、証明書ファイルの場所を決定します。

ssl.ca.location: certifi.where()
Copy

その他のクライアントの場合は、インストールパスを調べ、それをコードで指定します。

ssl.ca.location: '/usr/local/etc/openssl@1.1/cert.pem'
Copy

CentOS¶

次の方法で CA ルート証明書をインストールする必要がある場合があります。

sudo yum reinstall ca-certificates
Copy

これで、Kafka クライアントは証明書を見つけることができます。ただし、同じエラーが依然として表示される場合は、クライアントコードで ssl.ca.location プロパティを設定できます。プロデューサーとコンシューマーのコードファイルを両方とも編集し、プロデューサーおよびコンシューマーのプロパティ内に ssl.ca.location 構成パラメーターを追加します。その値は、次の例のように、ホスト上の適切な CA ルート証明書ファイルの場所に対応している必要があります。

ssl.ca.location: '/etc/ssl/certs/ca-bundle.crt'
Copy

Kafka クラスター¶

このチュートリアルは、どの環境の Kafka クラスターにも対応しています。
- Confluent Cloud 内
- ローカルホスト上
- 任意のリモート Kafka クラスター
Confluent Cloud で実行している場合は、API キーおよびシークレットで Confluent Cloud クラスターにアクセスできることが必要です。
- Confluent Cloud にサインアップし、プロモーションコード C50INTEG を使用した最初の 20 名のユーザーは、$50 相当を無料で使用できます（詳細）。
- Confluent Cloud で Kafka クラスター、資格情報、および ACL を自動的に作成する方法については、「Confluent Cloud 向け ccloud-stack ユーティリティ」を参照してください。

セットアップ¶

confluentinc/examples GitHub リポジトリのクローンを作成し、6.1.5-post ブランチをチェックアウトします。
```
git clone https://github.com/confluentinc/examples
cd examples
git checkout 6.1.5-post
```
Copy
Go のサンプルのディレクトリに変更します。
```
cd clients/cloud/go/
```
Copy
Kafka クラスターに接続するための構成パラメーターを含むローカルファイル（$HOME/.confluent/librdkafka.config など）を作成します。以下のテンプレートのいずれかをベースとして、クラスターへの接続情報でこのファイルをカスタマイズします。{{ BROKER_ENDPOINT }}、{{CLUSTER_API_KEY }}、および {{ CLUSTER_API_SECRET }} に値を代入します（この値を手作業で見つける方法については、「Configure Confluent Cloud Clients」を参照してください。または、Confluent Cloud 向け ccloud-stack ユーティリティを使用して値を自動作成します）。
- Confluent Cloud のテンプレート構成ファイル
```
# Kafka
bootstrap.servers={{ BROKER_ENDPOINT }}
security.protocol=SASL_SSL
sasl.mechanisms=PLAIN
sasl.username={{ CLUSTER_API_KEY }}
sasl.password={{ CLUSTER_API_SECRET }}
```
  Copy
- ローカルホストのテンプレート構成ファイル
```
# Kafka
bootstrap.servers=localhost:9092
```
  Copy

基本プロデューサーおよびコンシューマー¶

このサンプルでは、プロデューサーアプリケーションが Kafka クラスターのトピックに Kafka データを書き込みます。Kafka クラスターにトピックがまだ存在しない場合、プロデューサーアプリケーションは Kafka Admin Client API を使用してトピックを作成します。Kafka に書き込まれる各レコードには、ユーザー名（alice など）とカウント値を表す、json フォーマット（{"Count": 0} など）のキーがあります。コンシューマーアプリケーションは、同じ Kafka トピックを読み取り、各レコードの処理中にカウントのローリング合計を保持します。

レコードの生成¶

プロデューサーアプリケーションをビルドします。
```
go build producer.go
```
Copy
プロデューサーを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- トピック名
```
./producer -f $HOME/.confluent/librdkafka.config -t test1
```
Copy

プロデューサーがすべてのメッセージを送信したことを確認します。次のように表示されます。

Preparing to produce record: alice   {"Count": 0}
Preparing to produce record: alice   {"Count": 1}
Preparing to produce record: alice   {"Count": 2}
Preparing to produce record: alice   {"Count": 3}
Preparing to produce record: alice   {"Count": 4}
Preparing to produce record: alice   {"Count": 5}
Preparing to produce record: alice   {"Count": 6}
Preparing to produce record: alice   {"Count": 7}
Preparing to produce record: alice   {"Count": 8}
Preparing to produce record: alice   {"Count": 9}
Successfully produced record to topic test1 partition [0] @ offset 0
Successfully produced record to topic test1 partition [0] @ offset 1
Successfully produced record to topic test1 partition [0] @ offset 2
Successfully produced record to topic test1 partition [0] @ offset 3
Successfully produced record to topic test1 partition [0] @ offset 4
Successfully produced record to topic test1 partition [0] @ offset 5
Successfully produced record to topic test1 partition [0] @ offset 6
Successfully produced record to topic test1 partition [0] @ offset 7
Successfully produced record to topic test1 partition [0] @ offset 8
Successfully produced record to topic test1 partition [0] @ offset 9
10 messages were produced to topic test1!
Copy

プロデューサーコードを表示します。

レコードの消費¶

プロデューサーアプリケーションをビルドします。
```
go build consumer.go
```
Copy
コンシューマーを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- 手順 1 で使用したものと同じトピック名
```
./consumer -f $HOME/.confluent/librdkafka.config -t test1
```
Copy

コンシューマーがすべてのメッセージを受信したことを確認します。次のように表示されます。

...
Consumed record with key alice and value {"Count":0}, and updated total count to 0
Consumed record with key alice and value {"Count":1}, and updated total count to 1
Consumed record with key alice and value {"Count":2}, and updated total count to 3
Consumed record with key alice and value {"Count":3}, and updated total count to 6
Consumed record with key alice and value {"Count":4}, and updated total count to 10
Consumed record with key alice and value {"Count":5}, and updated total count to 15
Consumed record with key alice and value {"Count":6}, and updated total count to 21
Consumed record with key alice and value {"Count":7}, and updated total count to 28
Consumed record with key alice and value {"Count":8}, and updated total count to 36
Consumed record with key alice and value {"Count":9}, and updated total count to 45
...
Copy

コンシューマーコードを表示します。