Kafka インスタンスエンドポイントを使用したメッセージの送受信 - ApsaraMQ for Kafka

デフォルト、SSL、または SASL エンドポイントを介してアプリケーションを ApsaraMQ for Kafka インスタンスに接続し、メッセージを送受信します。

前提条件

JDK 1.8 以降をインストールします。詳細については、「JDK のインストール」をご参照ください。
Maven 2.5 以降をインストールします。詳細については、「Maven のインストール」をご参照ください。
ビルドツールをインストールします。

このトピックでは、例として IntelliJ IDEA Ultimate を使用します。
ApsaraMQ for Kafka インスタンスを購入してデプロイします。
- VPC インスタンス：同じ VPC 内からアクセス可能なデフォルトエンドポイントのみを提供します。
- インターネット/VPC インスタンス：デフォルトエンドポイントと SSL エンドポイントを提供します。インターネット経由または VPC からアクセスできます。
説明
- VPC インスタンスをインターネット/VPC インスタンスに、またはその逆に変換できます。詳細については、「インスタンス設定のアップグレードまたはダウングレード」をご参照ください。
- SASL エンドポイントはデフォルトで無効になっており、コンソールには表示されません。 SASL エンドポイントを使用するには、手動で有効にする必要があります。詳細については、「SASL ユーザーへの権限付与」をご参照ください。
- 各エンドポイントのユースケースの詳細については、「エンドポイントの比較」をご参照ください。

Java 依存関係のインストール

以下の依存関係が必要です。これらは kafka-java-demo フォルダーの pom.xml ファイルにすでに含まれているため、手動で追加する必要はありません。

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>2.6.0</version>
</dependency>
<dependency>
    <groupId>org.slf4j</groupId>
    <artifactId>slf4j-log4j12</artifactId>
    <version>1.7.6</version>
</dependency>

説明

クライアントライブラリのメジャーバージョンは、ご使用の ApsaraMQ for Kafka インスタンスのメジャーバージョンと一致させることをお勧めします。ご使用の ApsaraMQ for Kafka インスタンスのメジャーバージョンは、ApsaraMQ for Kafka コンソールの インスタンスの詳細 ページで確認できます。

設定

任意: SSL ルート証明書をダウンロード。 SSL エンドポイントを使用する場合は、この証明書をダウンロードする必要があります。
Aliware-kafka-demos に移動し、をクリックしてデモプロジェクトをダウンロードし、解凍します。
解凍したプロジェクトから kafka-java-demo フォルダーを IntelliJ IDEA にインポートします。
任意:インスタンスに接続するために SSL エンドポイントまたは SASL エンドポイントを使用する場合、kafka_client_jaas.conf ファイルを変更する必要があります。インスタンスエンドポイントの詳細については、「エンドポイントの比較」をご参照ください。
```
KafkaClient {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="xxxx"
  password="xxxx";
}; 
```
VPC インスタンスの場合、Kafka インスタンスには同じ VPC 内のリソースのみがアクセスでき、これにより安全でプライベートなデータ送信が保証されます。セキュリティを強化するために、ACL 機能を有効にして、SASL 認証後にのみメッセージが送信されるように設定できます。セキュリティ要件に基づいて、PLAIN メカニズムまたは SCRAM メカニズムを使用できます。詳細については、「ACL 機能の有効化」をご参照ください。

インターネット/VPC インスタンスの場合、インターネット経経由で送信されるメッセージは認証および暗号化される必要があります。 SSL トランスポート層では、PLAIN メカニズムを使用した SASL を使用する必要があります。 SASL_SSL プロトコルは、インターネット経由での平文送信を防ぎます。
例のユーザー名とパスワードは、インスタンスの SASL ユーザー名と SASL パスワードです。
- インターネット経由でアクセス可能なインスタンスの ACL 機能が無効になっている場合、デフォルトユーザーのユーザー名とパスワードは、ApsaraMQ for Kafka コンソールのインスタンスの詳細 ページにある設定情報 セクションから取得できます。
- インスタンスで ACL 機能が有効になっている場合は、SASL ユーザーが PLAIN タイプであり、メッセージを送受信する権限が付与されていることを確認してください。詳細については、「SASL ユーザーへの権限付与」をご参照ください。

kafka.properties 設定ファイルを変更します。

## ==============================共通パラメーター==============================
bootstrap.servers=xxxxxxxxxxxxxxxxxxxxx
topic=xxx
group.id=xxx
## =======================要件に応じて以下のパラメーターを設定========================
## SSL エンドポイント設定
ssl.truststore.location=/xxxx/only.4096.client.truststore.jks
## ssl.truststore.password の値は KafkaOnsClient で、変更できません。
ssl.truststore.password=KafkaOnsClient
## ホスト名の検証アルゴリズム。このパラメーターは空のままにし、変更しないでください。
ssl.endpoint.identification.algorithm=
java.security.auth.login.config=/xxxx/kafka_client_jaas.conf

パラメーター	説明
bootstrap.servers	エンドポイント。ApsaraMQ for Kafka コンソールのインスタンスの詳細ページのアクセスポイント情報セクションから取得できます。
トピック	トピックの名前。トピック名は、ApsaraMQ for Kafka コンソールのトピック管理ページで取得できます。
group.id	インスタンスのグループです。ApsaraMQ for Kafka コンソールの Group の管理ページで取得できます。説明このパラメーターはプロデューサーではオプションですが、コンシューマーでは必須です。
ssl.truststore.location	ダウンロードした SSL ルート証明書のローカルパスです。xxxx を実際のパスに置き換えてください。例: /home/ssl/only.4096.client.truststore.jks。重要このパラメーターは、デフォルトエンドポイントまたは SASL エンドポイントを使用する場合は不要です。 SSL エンドポイントを使用する場合は必須です。
ssl.truststore.password	トラストストアのパスワードです。値は KafkaOnsClient に固定されています。変更しないでください。
ssl.endpoint.identification.algorithm	ホスト名の検証アルゴリズムです。ホスト名の検証を無効にするには、これを空のままにします。
java.security.auth.login.config	JAAS 設定ファイルのパスです。デモプロジェクトから kafka_client_jaas.conf ファイルをローカルディレクトリに保存し、xxxx を実際のパスに置き換えます。例：/home/ssl/kafka_client_jaas.conf。重要このパラメーターは、デフォルトエンドポイントを使用する場合は不要です。 SSL エンドポイントまたは SASL エンドポイントを使用する場合は必須です。

メッセージの送信

KafkaProducerDemo.java をコンパイルして実行し、メッセージを送信します。

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
// SSL または SASL エンドポイントを使用する場合は、次の行をコメントアウトします。
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
/*
* SSL または SASL エンドポイントを使用する場合は、次の 2 行のコメントを解除します。
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaProducerDemo {

    public static void main(String args[]) {
          
       /*
        * SSL エンドポイントを使用する場合は、次の行のコメントを解除します。
        * JAAS 設定ファイルのパスを設定します。
        JavaKafkaConfigurer.configureSasl();
        */
         
       /*
        * PLAIN メカニズムを使用する SASL エンドポイントを使用する場合は、次の行のコメントを解除します。
        * JAAS 設定ファイルのパスを設定します。
        JavaKafkaConfigurer.configureSaslPlain();
        */
       
       /*
        * SCRAM メカニズムを使用する SASL エンドポイントを使用する場合は、次の行のコメントを解除します。
        * JAAS 設定ファイルのパスを設定します。
        JavaKafkaConfigurer.configureSaslScram();
        */

        // kafka.properties をロードします。
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // エンドポイントを設定します。コンソールからインスタンスのエンドポイントを取得してください。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
         
       /*
        * SSL エンドポイントを使用する場合は、次の 4 行のコメントを解除します。
        * SASL パスと同様に、このファイルは JAR ファイルにパッケージ化できません。
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        * トラストストアのパスワード。この値は変更しないでください。
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        * セキュリティプロトコル。SSL 接続の場合、これは SASL_SSL である必要があります。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        * SASL 認証方式。この値は変更しないでください。
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * PLAIN メカニズムを使用する SASL エンドポイントを使用する場合は、次の 2 行のコメントを解除します。
        * セキュリティプロトコル。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * PLAIN メカニズム。
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * SCRAM メカニズムを使用する SASL エンドポイントを使用する場合は、次の 2 行のコメントを解除します。
        * セキュリティプロトコル。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * SCRAM メカニズム。
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
        */

        // メッセージキーと値のシリアライザー。
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // リクエストの最大待機時間。
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        // クライアント側のリトライ回数。
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        // クライアント側のリトライ間隔。
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
         
       /*
        * SSL エンドポイントを使用する場合は、次の行のコメントを解除します。
        * 空の値はホスト名の検証を無効にします。
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        */

        // プロデューサーインスタンスを作成します。プロデューサーはスレッドセーフです。通常、1 つのプロセスには 1 つのプロデューサーインスタンスで十分です。
        // パフォーマンスを向上させるために、より多くのプロデューサーインスタンスを作成できますが、5 つ以下にすることを推奨します。
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        // Kafka メッセージを作成します。
        String topic = kafkaProperties.getProperty("topic"); // メッセージの送信先トピック。このトピックは、事前にコンソールで作成してください。
        String value = "this is the message's value"; // メッセージの内容。

        try {
            // メッセージをバッチで送信し、Future を収集するとパフォーマンスが向上しますが、バッチサイズを大きくしすぎないでください。
            List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
            for (int i =0; i < 100; i++) {
                // メッセージを送信し、Future オブジェクトを取得します。
                ProducerRecord<String, String> kafkaMessage =  new ProducerRecord<String, String>(topic, value + ": " + i);
                Future<RecordMetadata> metadataFuture = producer.send(kafkaMessage);
                futures.add(metadataFuture);

            }
            producer.flush();
            for (Future<RecordMetadata> future: futures) {
                // Future オブジェクトの結果を同期的に取得します。
                try {
                    RecordMetadata recordMetadata = future.get();
                    System.out.println("Produce ok:" + recordMetadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            // クライアント側でのリトライ後もメッセージの送信に失敗した場合は、アプリケーションでこのエラーを処理する必要があります。
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }
}

メッセージのサブスクライブ

次のいずれかの方法を選択して、メッセージをサブスクライブしてください。

単一コンシューマーによるサブスクリプション

KafkaConsumerDemo.java をコンパイルして実行し、メッセージを受信します。

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;


import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
/*
* SSL エンドポイントを使用する場合は、次の 3 行のコメントを解除します。SASL エンドポイントを使用する場合は、最初の 2 行のコメントを解除します。
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaConsumerDemo {

    public static void main(String args[]) {

        // JAAS 設定ファイルのパスを設定します。
        /*
         * SSL エンドポイントを使用する場合は、次の行のコメントを解除します。
        JavaKafkaConfigurer.configureSasl();
         */
                        
        /*
         * PLAIN メカニズムを使用する SASL エンドポイントを使用する場合は、次の行のコメントを解除します。
        JavaKafkaConfigurer.configureSaslPlain();
         */
                        
        /*
        * SCRAM メカニズムを使用する SASL エンドポイントを使用する場合は、次の行のコメントを解除します。
        JavaKafkaConfigurer.configureSaslScram();
        */

        // kafka.properties をロードします。
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // エンドポイントを設定します。コンソールからインスタンスのエンドポイントを取得してください。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

        // SSL エンドポイントを使用する場合は、次の行をコメントアウトします。
        // プルするデータ量とクライアントのバージョンに基づいてこの値を調整します。デフォルト値：30 秒。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * SSL エンドポイントを使用する場合は、次の 6 行のコメントを解除します。
         * SSL トラストストアファイルへのパス。このパスが kafka.properties ファイルで正しく設定されていることを確認してください。
         * SASL パスと同様に、このファイルは JAR ファイルにパッケージ化できません。
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * トラストストアのパスワード。この値は変更しないでください。
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * セキュリティプロトコル。SSL 接続の場合、これは SASL_SSL である必要があります。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * SASL 認証方式。この値は変更しないでください。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * 2 つのポーリング間の最大間隔。
         * コンシューマーがこの間隔内にハートビートを送信できなかった場合、ブローカーはコンシューマーを非アクティブとみなし、コンシューマーグループから削除して、リバランスをトリガーします。デフォルト値：30 秒。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         * リクエストごとにフェッチするデータ量。このパラメーターは、インターネット経由でインスタンスにアクセスする際のパフォーマンスに大きな影響を与える可能性があります。
         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
         */

        // PLAIN メカニズムを使用する SASL エンドポイントを使用する場合は、次の行をコメントアウトします。
        // プルするデータ量とクライアントのバージョンに基づいてこの値を調整します。デフォルト値：30 秒。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * PLAIN メカニズムを使用する SASL エンドポイントを使用する場合は、次の 3 行のコメントを解除します。
         * セキュリティプロトコル。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * PLAIN メカニズム。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * 2 つのポーリング間の最大間隔。
         * コンシューマーがこの間隔内にハートビートを送信できなかった場合、ブローカーはコンシューマーを非アクティブとみなし、コンシューマーグループから削除して、リバランスをトリガーします。デフォルト値：30 秒。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        // SCRAM メカニズムを使用する SASL エンドポイントを使用する場合は、次の行をコメントアウトします。
        // プルするデータ量とクライアントのバージョンに基づいてこの値を調整します。デフォルト値：30 秒。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * SCRAM メカニズムを使用する SASL エンドポイントを使用する場合は、次の 4 行のコメントを解除します。
         * セキュリティプロトコル。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * SCRAM メカニズム。
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         * 2 つのポーリング間の最大間隔。
         * コンシューマーがこの間隔内にハートビートを送信できなかった場合、ブローカーはコンシューマーを非アクティブとみなし、コンシューマーグループから削除して、リバランスをトリガーします。デフォルト値：30 秒。
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        // 1 回の poll() 呼び出しで返されるレコードの最大数。
        // この値を大きな数値に設定しないでください。大量のデータをポーリングしても、次のポーリングまでにデータを消費できない場合、リバランスがトリガーされ、遅延が発生する可能性があります。
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        // メッセージキーと値のデシリアライザー。
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 現在のコンシューマーインスタンスが属するコンシューマーグループ。コンシューマーグループは、事前にコンソールで作成してください。
        // 同じグループ内のコンシューマーインスタンスは、負荷分散方式でメッセージを消費します。
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
        
        // SSL エンドポイントを使用する場合は、次の行のコメントを解除します。
        // 空の値はホスト名の検証を無効にします。
        //props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        // コンシューマーインスタンスを作成します。
        KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
        // トピックをサブスクライブします。複数のトピックをサブスクライブできます。
        // コンシューマーインスタンスの GROUP_ID_CONFIG 値が同じ場合は、同じトピックをサブスクライブするように設定することを推奨します。
        List<String> subscribedTopics =  new ArrayList<String>();
        
        // 単一のトピックのみをサブスクライブする場合は、次の 5 行をコメントアウトし、6 行目のコメントを解除します。
        // 複数のトピックをサブスクライブする場合は、ここに追加します。
        // 各トピックは、事前にコンソールで作成してください。
        String topicStr = kafkaProperties.getProperty("topic");
        String[] topics = topicStr.split(",");
        for (String topic: topics) {
            subscribedTopics.add(topic.trim());
        }
        //subscribedTopics.add(kafkaProperties.getProperty("topic"));
        consumer.subscribe(subscribedTopics);

        // ループ内でメッセージを消費します。
        while (true){
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                // ポーリングされたデータは、次のポーリングの前に消費する必要があります。合計時間は SESSION_TIMEOUT_MS_CONFIG の値を超えることはできません。
                // メッセージを消費し、結果を非同期に返すために、別のスレッドプールを作成することを推奨します。
                for (ConsumerRecord<String, String> record : records) {
                    System.out.println(String.format("Consume partition:%d offset:%d", record.partition(), record.offset()));
                }
            } catch (Exception e) {
                try {
                    Thread.sleep(1000);
                } catch (Throwable ignore) {

                }
          
                e.printStackTrace();
            }
        }
    }
}

複数コンシューマーによるサブスクリプション

KafkaMultiConsumerDemo.java をコンパイルして実行し、メッセージを消費します。

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
// SSL または SASL エンドポイントを使用する場合は、次の行のコメントを解除します。

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
/*
* SSL エンドポイントを使用する場合は、次の 3 行のコメントを解除します。SASL エンドポイントを使用する場合は、最初の 2 行のコメントを解除します。
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/
import org.apache.kafka.common.errors.WakeupException;

/**
 * このデモでは、単一のプロセスで複数のコンシューマーを起動して、トピックから同時にメッセージを消費する方法を示します。
 * コンシューマーの総数が、サブスクライブしたトピックのパーティションの総数を超えないようにしてください。
 */
public class KafkaMultiConsumerDemo {

    public static void main(String args[]) throws InterruptedException {
        
        // JAAS 設定ファイルのパスを設定します。
        /* 
         * SSL エンドポイントを使用する場合は、次の行のコメントを解除します。
         JavaKafkaConfigurer.configureSasl();
         */
                            
        /* 
         * PLAIN メカニズムを使用する SASL エンドポイントを使用する場合は、次の行のコメントを解除します。
         JavaKafkaConfigurer.configureSaslPlain(); 
         */
                            
        /* 
         * SCRAM メカニズムを使用する SASL エンドポイントを使用する場合は、次の行のコメントを解除します。
         JavaKafkaConfigurer.configureSaslScram();
         */


        // kafka.properties をロードします。
        Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // エンドポイントを設定します。コンソールからインスタンスのエンドポイントを取得してください。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        
        /*
         * SSL エンドポイントを使用する場合は、次の 4 行のコメントを解除します。
         * SASL パスと同様に、このファイルは JAR ファイルにパッケージ化できません。
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * トラストストアのパスワード。この値は変更しないでください。
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * セキュリティプロトコル。SSL 接続の場合、これは SASL_SSL である必要があります。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * SASL 認証方式。この値は変更しないでください。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */
        
        /*
         * PLAIN メカニズムを使用する SASL エンドポイントを使用する場合は、次の 2 行のコメントを解除します。
         * セキュリティプロトコル。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * PLAIN メカニズム。
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */

        /* 
         * SCRAM メカニズムを使用する SASL エンドポイントを使用する場合は、次の 2 行のコメントを解除します。
         * セキュリティプロトコル。
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * SCRAM メカニズム。
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         */

        // 2 つのポーリング間の最大間隔。
        // コンシューマーがこの間隔内にハートビートを送信できなかった場合、ブローカーはコンシューマーを非アクティブとみなし、コンシューマーグループから削除して、リバランスをトリガーします。デフォルト値：30 秒。
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        // 1 回の poll() 呼び出しで返されるレコードの最大数。
        // この値を大きな数値に設定しないでください。大量のデータをポーリングしても、次のポーリングまでにデータを消費できない場合、リバランスがトリガーされ、遅延が発生する可能性があります。
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        // メッセージキーと値のデシリアライザー。
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        // 現在のコンシューマーインスタンスが属するコンシューマーグループ。コンシューマーグループは、事前にコンソールで作成してください。
        // 同じグループ内のコンシューマーインスタンスは、負荷分散方式でメッセージを消費します。
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));

        /* 
         * SSL エンドポイントを使用する場合は、次の行のコメントを解除します。
         * 空の値はホスト名の検証を無効にします。
         props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
         */

        int consumerNum = 2;
        Thread[] consumerThreads = new Thread[consumerNum];
        for (int i = 0; i < consumerNum; i++) {
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

            List<String> subscribedTopics = new ArrayList<String>();
            subscribedTopics.add(kafkaProperties.getProperty("topic"));
            consumer.subscribe(subscribedTopics);

            KafkaConsumerRunner kafkaConsumerRunner = new KafkaConsumerRunner(consumer);
            consumerThreads[i] = new Thread(kafkaConsumerRunner);
        }

        for (int i = 0; i < consumerNum; i++) {
            consumerThreads[i].start();
        }

        for (int i = 0; i < consumerNum; i++) {
            consumerThreads[i].join();
        }
    }

    static class KafkaConsumerRunner implements Runnable {
        private final AtomicBoolean closed = new AtomicBoolean(false);
        private final KafkaConsumer consumer;

        KafkaConsumerRunner(KafkaConsumer consumer) {
            this.consumer = consumer;
        }

        @Override
        public void run() {
            try {
                while (!closed.get()) {
                    try {
                        ConsumerRecords<String, String> records = consumer.poll(1000);
                        // ポーリングされたデータは、次のポーリングの前に消費する必要があります。合計時間は SESSION_TIMEOUT_MS_CONFIG の値を超えることはできません。
                        for (ConsumerRecord<String, String> record : records) {
                            System.out.println(String.format("Thread:%s Consume partition:%d offset:%d", Thread.currentThread().getName(), record.partition(), record.offset()));
                        }
                    } catch (Exception e) {
                        try {
                            Thread.sleep(1000);
                        } catch (Throwable ignore) {

                        }
                        e.printStackTrace();
                    }
                }
            } catch (WakeupException e) {
                // コンシューマーがクローズされている場合は、例外を無視します。
                if (!closed.get()) {
                    throw e;
                }
            } finally {
                consumer.close();
            }
        }
        // 別のスレッドから呼び出すことができるシャットダウンフック。
        public void shutdown() {
            closed.set(true);
            consumer.wakeup();
        }
    }
}

よくある質問

SASL_SSL 証明書の設定

「設定」セクションの手順 1 の URL から SSL 証明書をダウンロードします。次に、kafka.properties ファイルで、ssl.truststore.location を証明書のローカルパスに設定します。

カスタム SSL 証明書の使用

いいえ。 ApsaraMQ for Kafka が提供する SSL 証明書を使用する必要があります。

ApsaraMQ for Kafka:インスタンスエンドポイントを使用したメッセージの送受信