ApsaraMQ for Kafka:Envoyer et recevoir des messages via un endpoint d'instance

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;
/*
* If you use an SSL endpoint, uncomment the following three lines. If you use a SASL endpoint, uncomment the first two lines.
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[]) {

        // Set the path of the JAAS configuration file.
        /*
         * If you use an SSL endpoint, uncomment the following line.
        JavaKafkaConfigurer.configureSasl();
         */
                        
        /*
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following line.
        JavaKafkaConfigurer.configureSaslPlain();
         */
                        
        /*
        * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following line.
        JavaKafkaConfigurer.configureSaslScram();
        */

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

        Properties props = new Properties();
        // Set the endpoint. Obtain the instance endpoint from the console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));

        // If you use an SSL endpoint, comment out the following line.
        // Adjust this value based on the amount of data to pull and the client version. Default value: 30s.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * If you use an SSL endpoint, uncomment the following six lines.
         * The path to the SSL truststore file. Ensure this path is correctly configured in your kafka.properties file.
         * Similar to the SASL path, this file cannot be packaged into a JAR file.
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * The password for the truststore. Do not change this value.
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * The security protocol. For SSL connections, this must be SASL_SSL.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * The SASL authentication method. Do not change this value.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * The maximum interval allowed between two polls.
         * If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         * The amount of data to fetch per request. This parameter can significantly affect performance when accessing the instance over the Internet.
         props.put(ConsumerConfig.MAX_PARTITION_FETCH_BYTES_CONFIG, 32000);
         props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 32000);
         */

        // If you use a SASL endpoint with the PLAIN mechanism, comment out the following line.
        // Adjust this value based on the amount of data to pull and the client version. Default value: 30s.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following three lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The PLAIN mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         * The maximum interval allowed between two polls.
         * If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        // If you use a SASL endpoint with the SCRAM mechanism, comment out the following line.
        // Adjust this value based on the amount of data to pull and the client version. Default value: 30s
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        /*
         * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following four lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The SCRAM mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         * The maximum interval allowed between two polls.
         * If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
         props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
         */

        // The maximum number of records to return in a single call to poll().
        // Do not set this value to a large number. If you poll a large amount of data but cannot consume the data before the next poll, a rebalance is triggered, which may cause lags.
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        // The deserializers for message keys and values.
        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");
        // The consumer group to which the current consumer instance belongs. You must create the consumer group in the console first.
        // Consumer instances in the same group consume messages in a load-balanced manner.
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));
        
        // If you use an SSL endpoint, uncomment the following line.
        // An empty value disables hostname verification.
        //props.put(SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");

        // Create a consumer instance.
        KafkaConsumer<String, String> consumer = new org.apache.kafka.clients.consumer.KafkaConsumer<String, String>(props);
        // Subscribe to topics. You can subscribe to multiple topics.
        // If the consumer instances have the same GROUP_ID_CONFIG value, we recommend that you configure them to subscribe to the same topics.
        List<String> subscribedTopics =  new ArrayList<String>();
        
        // If you use an SSL endpoint, comment out the first five lines and uncomment the sixth line.
        // If you want to subscribe to multiple topics, add them here.
        // You must create each topic in the console first.
        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);

        // Consume messages in a loop.
        while (true){
            try {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                // The polled data must be consumed before the next poll. The total time cannot exceed the value of SESSION_TIMEOUT_MS_CONFIG.
                // We recommend that you create a separate thread pool to consume messages and asynchronously return the results.
                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();
            }
        }
    }
}

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.atomic.AtomicBoolean;
// If you use an SSL or SASL endpoint, uncomment the following line.

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;
/*
* If you use an SSL endpoint, uncomment the following three lines. If you use a SASL endpoint, uncomment the first two lines.
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;

/**
 * This demo shows how to start multiple consumers in a single process to consume messages from a topic at the same time.
 * Make sure that the total number of consumers does not exceed the total number of partitions for the subscribed topic.
 */
public class KafkaMultiConsumerDemo {

    public static void main(String args[]) throws InterruptedException {
        
        // Set the path of the JAAS configuration file.
        /* 
         * If you use an SSL endpoint, uncomment the following line.
         JavaKafkaConfigurer.configureSasl();
         */
                            
        /* 
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following line.
         JavaKafkaConfigurer.configureSaslPlain(); 
         */
                            
        /* 
         * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following line.
         JavaKafkaConfigurer.configureSaslScram();
         */

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

        Properties props = new Properties();
        // Set the endpoint. Obtain the instance endpoint from the console.
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaProperties.getProperty("bootstrap.servers"));
        
        /*
         * If you use an SSL endpoint, uncomment the following four lines.
         * Similar to the SASL path, this file cannot be packaged into a JAR file.
         props.put(SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, kafkaProperties.getProperty("ssl.truststore.location"));
         * The password for the truststore. Do not change this value.
         props.put(SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, "KafkaOnsClient");
         * The security protocol. For SSL connections, this must be SASL_SSL.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_SSL");
         * The SASL authentication method. Do not change this value.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */
        
        /*
         * If you use a SASL endpoint with the PLAIN mechanism, uncomment the following two lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The PLAIN mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "PLAIN");
         */

        /* 
         * If you use a SASL endpoint with the SCRAM mechanism, uncomment the following two lines.
         * The security protocol.
         props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
         * The SCRAM mechanism.
         props.put(SaslConfigs.SASL_MECHANISM, "SCRAM-SHA-256");
         */

        // The maximum interval allowed between two polls.
        // If a consumer fails to send a heartbeat within this interval, the broker considers the consumer inactive, removes it from the consumer group, and triggers a rebalance. Default value: 30s.
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 30000);
        // The maximum number of records to return in a single call to poll().
        // Do not set this value to a large number. If you poll a large amount of data but cannot consume the data before the next poll, a rebalance is triggered, which may cause lags.
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 30);
        // The deserializers for message keys and values.
        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");
        // The consumer group to which the current consumer instance belongs. You must create the consumer group in the console first.
        // Consumer instances in the same group consume messages in a load-balanced manner.
        props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaProperties.getProperty("group.id"));

        /* 
         * If you use an SSL endpoint, uncomment the following line.
         * An empty value disables hostname verification.
         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);
                        // The polled data must be consumed before the next poll. The total time cannot exceed the value of 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) {
                // Ignore the exception if the consumer is closed.
                if (!closed.get()) {
                    throw e;
                }
            } finally {
                consumer.close();
            }
        }
        // A shutdown hook that can be called by another thread.
        public void shutdown() {
            closed.set(true);
            consumer.wakeup();
        }
    }
}