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

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

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

前提条件¶

クライアント¶

Java 8 以降（Clojure 1.10 では、Java 8 または Java 11 の使用を推奨）
デモのコンパイルと実行を行うための Leiningen ツール

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
Clojure のサンプルのディレクトリに変更します。
```
cd clients/cloud/clojure/
```
Copy
Kafka クラスターに接続するための構成パラメーターを含むローカルファイル（$HOME/.confluent/java.config など）を作成します。以下のテンプレートのいずれかをベースとして、クラスターへの接続情報でこのファイルをカスタマイズします。{{ BROKER_ENDPOINT }}、{{CLUSTER_API_KEY }}、および {{ CLUSTER_API_SECRET }} に値を代入します（この値を手作業で見つける方法については、「Configure Confluent Cloud Clients」を参照してください。または、Confluent Cloud 向け ccloud-stack ユーティリティを使用して値を自動作成します）。
- Confluent Cloud のテンプレート構成ファイル
```
# Required connection configs for Kafka producer, consumer, and admin
bootstrap.servers={{ BROKER_ENDPOINT }}
security.protocol=SASL_SSL
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='{{ CLUSTER_API_KEY }}' password='{{ CLUSTER_API_SECRET }}';
sasl.mechanism=PLAIN
# Required for correctness in Apache Kafka clients prior to 2.6
client.dns.lookup=use_all_dns_ips

# Best practice for Kafka producer to prevent data loss 
acks=all
```
  Copy
- ローカルホストのテンプレート構成ファイル
```
# Kafka
bootstrap.servers=localhost:9092
```
  Copy

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

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

レコードの生成¶

プロデューサーを実行して、次の引数を渡します。

Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
トピック名

lein producer $HOME/.confluent/java.config test1
Copy

次のように表示されます。

…
Producing record: alice     {"count":0}
Producing record: alice     {"count":1}
Producing record: alice     {"count":2}
Producing record: alice     {"count":3}
Producing record: alice     {"count":4}
Produced record to topic test1 partition [0] @ offest 0
Produced record to topic test1 partition [0] @ offest 1
Produced record to topic test1 partition [0] @ offest 2
Produced record to topic test1 partition [0] @ offest 3
Produced record to topic test1 partition [0] @ offest 4
Producing record: alice     {"count":5}
Producing record: alice     {"count":6}
Producing record: alice     {"count":7}
Producing record: alice     {"count":8}
Producing record: alice     {"count":9}
Produced record to topic test1 partition [0] @ offest 5
Produced record to topic test1 partition [0] @ offest 6
Produced record to topic test1 partition [0] @ offest 7
Produced record to topic test1 partition [0] @ offest 8
Produced record to topic test1 partition [0] @ offest 9
10 messages were produced to topic test1!
Copy

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

レコードの消費¶

コンシューマーを実行して、次の引数を渡します。
- Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
- 前の手順で使用したトピック名
```
lein consumer $HOME/.confluent/java.config test1
```
Copy

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

…
Waiting for message in KafkaConsumer.poll
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
Waiting for message in KafkaConsumer.poll
…
Copy

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