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存取點執行個體，請取消以下兩行代碼的注釋。
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的存取點。
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
         
       /*
        * 如果是SSL存取點執行個體，請取消以下四行代碼的注釋。
        * 與sasl路徑類似，該檔案也不能被打包到jar中。
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        * 根憑證store的密碼，保持不變。
        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機制執行個體，請取消以下兩行代碼的注釋。
        * 接入協議。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * Plain方式。
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * 如果是SASL存取點SCRAM機制執行個體，請取消以下兩行代碼的注釋。
        * 接入協議。
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * SCRAM方式。
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
        */

        //雲訊息佇列 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);
        //設定用戶端內部稍候再試。
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
         
       /*
        * 如果是SSL存取點執行個體或，請取消以下一行代碼的注釋。
        * Hostname校正改成空。
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        */

        //構造Producer對象，注意，該對象是安全執行緒的，一般來說，一個進程內一個Producer對象即可。
        //如果想提高效能，可以多構造幾個對象，但不要太多，最好不要超過5個。
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        //構造一個雲訊息佇列 Kafka 版訊息。
        String topic = kafkaProperties.getProperty("topic"); //訊息所屬的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();
        }
    }
}