1. Home
  2. Platform
  3. Kafka スタートガイド
  4. ビデオ、デモ、および参考文献
  5. Confluent Cloud のサンプルのインデックス
  6. Apache Kafka® のサンプルコード

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

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

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

前提条件

クライアント

  • Java 1.8 以降（デモアプリケーションの実行用）

Kafka クラスター

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

セットアップ

  1. confluentinc/examples GitHub リポジトリのクローンを作成し、6.1.5-post ブランチをチェックアウトします。

    git clone https://github.com/confluentinc/examples
cd examples
git checkout 6.1.5-post
    Copy

  2. Kotlin のサンプルのディレクトリに変更します。

    cd clients/cloud/kotlin/
    Copy

  3. 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 トピックを読み取り、各レコードの処理中にカウントのローリング合計を保持します。

レコードの生成

  1. サンプルクライアントをビルドします。

    ./gradlew clean build
    Copy

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

    • Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
    • トピック名
    ./gradlew runApp -PmainClass="io.confluent.examples.clients.cloud.ProducerExample" \
     -PconfigPath="$HOME/.confluent/java.config" \
     -Ptopic="test1"
    Copy

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

    ...
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}
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] @ offset 0
Produced record to topic test1 partition [0] @ offset 1
Produced record to topic test1 partition [0] @ offset 2
Produced record to topic test1 partition [0] @ offset 3
Produced record to topic test1 partition [0] @ offset 4
Produced record to topic test1 partition [0] @ offset 5
Produced record to topic test1 partition [0] @ offset 6
Produced record to topic test1 partition [0] @ offset 7
Produced record to topic test1 partition [0] @ offset 8
Produced record to topic test1 partition [0] @ offset 9
10 messages were produced to topic test1
...
    Copy

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

レコードの消費

  1. コンシューマーを実行して、次の引数を渡します。

    • Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
    • 前の手順で使用したものと同じトピック名
    ./gradlew runApp -PmainClass="io.confluent.examples.clients.cloud.ConsumerExample"\
     -PconfigPath="$HOME/.confluent/java.config"\
     -Ptopic="test1"
    Copy

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

    ...
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

  3. 入力し終わったら、Ctrl + C キーを押します（"+" はキーを同時に押すことを意味します）。

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

Kafka Streams

  1. Kafka Streams アプリケーションを実行して、次の引数を渡します。

    • Kafka クラスターに接続するための構成パラメーターを含むローカルファイル
    • 前の手順で使用したものと同じトピック名
    ./gradlew runApp -PmainClass="io.confluent.examples.clients.cloud.StreamsExample" \
     -PconfigPath="$HOME/.confluent/java.config" \
     -Ptopic="test1"
    Copy

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

    ...
[Consumed record]: alice, 0
[Consumed record]: alice, 1
[Consumed record]: alice, 2
[Consumed record]: alice, 3
[Consumed record]: alice, 4
[Consumed record]: alice, 5
[Consumed record]: alice, 6
[Consumed record]: alice, 7
[Consumed record]: alice, 8
[Consumed record]: alice, 9
...
[Running count]: alice, 0
[Running count]: alice, 1
[Running count]: alice, 3
[Running count]: alice, 6
[Running count]: alice, 10
[Running count]: alice, 15
[Running count]: alice, 21
[Running count]: alice, 28
[Running count]: alice, 36
[Running count]: alice, 45
...
    Copy

  3. 入力し終わったら、Ctrl + C キーを押します（"+" はキーを同時に押すことを意味します）。

  4. Kafka Streams コード を表示します。