ApsaraMQ for Kafka:SSLエンドポイントとPLAIN認証を使用してメッセージを送受信する

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>

次のいずれかの方法でメッセージをコンシュームできます。

• 単一のコンシューマーを使用してメッセージをコンシュームする

  1. 次のコードを含む KafkaConsumerDemo.java という名前の単一コンシューマープログラムを作成します。

     import java.util.ArrayList;
     import java.util.List;
     import java.util.Properties;
     import org.apache.kafka.clients.CommonClientConfigs;
     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;
     import org.apache.kafka.common.config.SaslConfigs;
     import org.apache.kafka.common.config.SslConfigs;
     
     public class KafkaConsumerDemo {
     
         public static void main(String args[]) {
             // JAAS設定ファイルのパスを指定します。
             JavaKafkaConfigurer.configureSasl();
     
             // kafka.propertiesファイルをロードします。
             Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();
     
             Properties props = new Properties();
             // エンドポイントを指定します。Message Queue for Apache Kafkaコンソールで対応するインスタンスのSASLエンドポイントを取得します。
             props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
             // SSLルート証明書のパスを指定します。XXXを実際のパスに置き換えます。
             // 証明書ファイルをJARパッケージに圧縮しないでください。
             props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
             // ルート証明書ストアのトラストストアのパスワードを指定します。デフォルト値を使用します。
             props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
             // アクセスプロトコルを指定します。値を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);
             // 一度にポーリングできるメッセージの最大数を設定します。
             // このパラメーターを過度に大きな値に設定しないでください。ポーリングされたメッセージが次のポーリングサイクルが始まる前にすべてコンシュームされない場合、ロードバランシングがトリガーされ、パフォーマンスが低下する可能性があります。
             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");
             // 現在のコンシューマーのコンシューマーグループを設定します。Message Queue for Apache Kafkaコンソールでコンシューマーグループを作成する必要があります。
             // コンシューマーグループ内のコンシューマーは、ロードバランシングモードでメッセージをコンシュームします。
             props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
             // ホスト名検証のアルゴリズムを空の値に設定します。
             props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
     
             // コンシューマーオブジェクトを作成します。
             KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
             // コンシューマーグループがサブスクライブする1つ以上のトピックを指定します。
             // 同じGROUP_ID_CONFIG値を持つコンシューマーを構成して、同じトピックにサブスクライブすることをお勧めします。
             List<String> subscribedTopics =  new ArrayList<String>();
             // 複数のトピックにサブスクライブする場合は、ここにトピックを追加します。
             // Message Queue for Apache Kafkaコンソールで事前にトピックを作成する必要があります。
             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();
                 }
             }
         }
     }

  2. KafkaConsumerDemo.java をコンパイルして実行し、メッセージをコンシュームします。

• 複数のコンシューマーを使用してメッセージをコンシュームする

  1. 次のコードを含む KafkaMultiConsumerDemo.java という名前のマルチコンシューマープログラムを作成します。

     import java.util.ArrayList;
     import java.util.List;
     import java.util.Properties;
     import java.util.concurrent.atomic.AtomicBoolean;
     import org.apache.kafka.clients.CommonClientConfigs;
     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;
     import org.apache.kafka.common.config.SaslConfigs;
     import org.apache.kafka.common.config.SslConfigs;
     import org.apache.kafka.common.errors.WakeupException;
     
     /**
      * このチュートリアルでは、1つのプロセスで複数のコンシューマーを使用して同時にメッセージをコンシュームする方法を示します。
      * 環境内のコンシューマーの総数が、コンシューマーがサブスクライブするトピックのパーティション数を超えないようにしてください。
      */
     public class KafkaMultiConsumerDemo {
     
         public static void main(String args[]) throws InterruptedException {
             // JAAS設定ファイルのパスを指定します。
             JavaKafkaConfigurer.configureSasl();
     
             // kafka.propertiesファイルをロードします。
             Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();
     
             Properties props = new Properties();
             // エンドポイントを指定します。Message Queue for Apache Kafkaコンソールで対応するインスタンスのSASLエンドポイントを取得します。
             props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
             // SSLルート証明書のパスを指定します。XXXを実際のパスに置き換えます。
             // 証明書ファイルをJARパッケージに圧縮しないでください。
             props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
             // ルート証明書ストアのトラストストアのパスワードを指定します。デフォルト値を使用します。
             props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
             // アクセスプロトコルを指定します。値を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_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");
             // 現在のコンシューマーのコンシューマーグループを設定します。Message Queue for Apache Kafkaコンソールでコンシューマーグループを作成する必要があります。
             // コンシューマーグループ内のコンシューマーは、ロードバランシングモードでメッセージをコンシュームします。
             props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
             // コンシューマーオブジェクトを作成します。
     
             // ホスト名検証のアルゴリズムを空の値に設定します。
             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();
             }
         }
     }

  2. KafkaMultiConsumerDemo.java をコンパイルして実行し、メッセージをコンシュームします。