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

次のサンプルコードは、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();
        */
         
       /*
        * SASL エンドポイントの PLAIN メカニズムを使用してインスタンスにアクセスする場合は、次の行のコードのコメントを解除します。
        JAAS 構成ファイルのパスを指定します。
        JavaKafkaConfigurer.configureSaslPlain();
        */
       
       /*
        * SASL エンドポイントの SCRAM メカニズムを使用してインスタンスにアクセスする場合は、次の行のコードのコメントを解除します。
        JAAS 構成ファイルのパスを指定します。
        JavaKafkaConfigurer.configureSaslScram();
        */

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

        Properties props = new Properties();
        // エンドポイントを指定します。Topic へのアクセスに使用するエンドポイントは、ApsaraMQ for Kafka コンソールの [インスタンス詳細] ページで取得できます。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
         
       /*
        * SSL エンドポイントを使用してインスタンスにアクセスする場合は、次の 4 行のコードのコメントを解除します。
        * ファイルを 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");
        */

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

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

        // ApsaraMQ for Kafka でメッセージをシリアル化するために使用されるメソッド。
        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);
        // クライアントでのメッセージの 2 回の連続したリトライの間隔。
        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);

        // ApsaraMQ for Kafka メッセージを構築します。
        String topic = kafkaProperties.getProperty("topic"); // メッセージが属する Topic。ApsaraMQ for Kafka コンソールで作成した Topic を入力します。
        String value = "this is the message's value"; // メッセージの内容。

        try {
            // 複数の future オブジェクトを同時に取得します。これにより効率が向上します。ただし、一度に多数の 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();
        }
    }
}