ApsaraMQ for Kafka：Gunakan titik akhir instans untuk mengirim dan menerima pesan

Kode contoh berikut menunjukkan cara mengompilasi dan menjalankan KafkaProducerDemo.java untuk mengirim pesan:

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;
// Jika Anda menggunakan titik akhir SSL atau titik akhir SASL untuk mengakses instans, komentari baris pertama kode berikut: 
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;
/*
* Jika Anda menggunakan titik akhir SSL atau titik akhir SASL untuk mengakses instans, hapus komentar dua baris kode berikut: 
import org.apache.kafka.common.config.SaslConfigs;
import org.apache.kafka.common.config.SslConfigs;
*/

public class KafkaProducerDemo {

    public static void main(String args[]) {
          
       /*
        * Jika Anda menggunakan titik akhir SSL untuk mengakses instans, hapus komentar baris kode berikut. 
        Tentukan jalur file konfigurasi JAAS. 
        JavaKafkaConfigurer.configureSasl();
        */
         
       /*
        * Jika Anda menggunakan mekanisme PLAIN dari titik akhir SASL untuk mengakses instans, hapus komentar baris kode berikut. 
        Tentukan jalur file konfigurasi JAAS. 
        JavaKafkaConfigurer.configureSaslPlain();
        */
       
       /*
        * Jika Anda menggunakan mekanisme SCRAM dari titik akhir SASL untuk mengakses instans, hapus komentar baris kode berikut. 
        Tentukan jalur file konfigurasi JAAS. 
        JavaKafkaConfigurer.configureSaslScram();
        */

        // Muat file kafka.properties. 
        Properties kafkaProperties =  JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();
        // Tentukan titik akhir. Anda dapat memperoleh titik akhir yang digunakan untuk mengakses topik di halaman Instance Details di konsol ApsaraMQ for Kafka. 
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
         
       /*
        * Jika Anda menggunakan titik akhir SSL untuk mengakses instans, hapus komentar empat baris kode berikut. 
        * Jangan kompres file ke dalam paket JAR. 
        props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
        * Kata sandi truststore dalam sertifikat root. Gunakan nilai default. 
        props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
        * Protokol akses. Atur parameter ini menjadi SASL_SSL. 
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
        * Metode autentikasi SASL. Gunakan nilai default. 
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * Jika Anda menggunakan mekanisme PLAIN dari titik akhir SASL untuk mengakses instans, hapus komentar dua baris kode berikut. 
        * Protokol akses. 
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * Mekanisme PLAIN. 
        props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
        */

       /*
        * Jika Anda menggunakan mekanisme SCRAM dari titik akhir SASL untuk mengakses instans, hapus komentar dua baris kode berikut. 
        * Protokol akses. 
        props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
        * Mekanisme SCRAM. 
        props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
        */

        // Metode yang digunakan untuk serialisasi pesan di 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");
        // Waktu tunggu maksimum untuk permintaan. 
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);
        // Jumlah maksimum percobaan ulang pesan di sisi klien. 
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        // Interval antara dua percobaan ulang berturut-turut untuk pesan di sisi klien. 
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);
         
       /*
        * Jika Anda menggunakan titik akhir SSL untuk mengakses instans, hapus komentar baris kode berikut. 
        * Atur algoritma verifikasi hostname menjadi nilai kosong. 
        props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
        */

        // Buat objek producer yang aman untuk thread. Buat satu objek producer untuk satu proses. 
        // Untuk meningkatkan kinerja, Anda dapat membuat beberapa objek producer. Kami menyarankan agar Anda membuat maksimal lima objek producer. 
        KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);

        // Buat pesan ApsaraMQ for Kafka. 
        String topic = kafkaProperties.getProperty("topic"); // Topik tempat pesan tersebut berada. Masukkan topik yang telah Anda buat di konsol ApsaraMQ for Kafka. 
        String value = "this is the message's value"; // Isi pesan. 

        try {
            // Peroleh beberapa objek future sekaligus. Hal ini membantu meningkatkan efisiensi. Namun, jangan peroleh terlalu banyak objek future sekaligus. 
            List<Future<RecordMetadata>> futures = new ArrayList<Future<RecordMetadata>>(128);
            for (int i =0; i < 100; i++) {
                // Kirim pesan dan peroleh objek 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) {
                // Peroleh hasil objek future secara sinkron. 
                try {
                    RecordMetadata recordMetadata = future.get();
                    System.out.println("Produce ok:" + recordMetadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            // Jika pesan tetap gagal dikirim setelah mencapai jumlah maksimum percobaan ulang, lakukan pemecahan masalah. 
            System.out.println("error occurred");
            e.printStackTrace();
        }
    }
}