ApsaraMQ for Kafka:Send and receive messages through the default endpoint

<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>

bootstrap.servers=<your-default-endpoint>
topic=<your-topic-name>
group.id=<your-group-id>

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
import java.util.concurrent.Future;

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;

public class KafkaProducerDemo {

    public static void main(String[] args) {
        // Load kafka.properties from the demo project.
        Properties kafkaProperties = JavaKafkaConfigurer.getKafkaProperties();

        Properties props = new Properties();

        // Default endpoint (bootstrap.servers from kafka.properties).
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
                kafkaProperties.getProperty("bootstrap.servers"));

        // Serialize both keys and values as strings.
        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");

        // Maximum wait time for a request: 30 seconds.
        props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 30 * 1000);

        // Retry up to 5 times on transient failures, with a 3-second backoff.
        props.put(ProducerConfig.RETRIES_CONFIG, 5);
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 3000);

        // Create a thread-safe producer.
        // One producer per process is sufficient. For higher throughput, use up to five.
        KafkaProducer<String, String> producer = new KafkaProducer<>(props);

        String topic = kafkaProperties.getProperty("topic");

        try {
            List<Future<RecordMetadata>> futures = new ArrayList<>(128);

            for (int i = 0; i < 100; i++) {
                ProducerRecord<String, String> record =
                        new ProducerRecord<>(topic, "message value: " + i);
                futures.add(producer.send(record));
            }

            producer.flush();

            for (Future<RecordMetadata> future : futures) {
                try {
                    RecordMetadata metadata = future.get();
                    System.out.println("Produce ok: " + metadata.toString());
                } catch (Throwable t) {
                    t.printStackTrace();
                }
            }
        } catch (Exception e) {
            // Troubleshoot if messages still fail after retries.
            System.out.println("Failed to send messages.");
            e.printStackTrace();
        }
    }
}

Comprobación de la salida del productor

Tras una ejecución exitosa, la consola imprime una línea Produce ok por cada mensaje, con el topic, la partición y el offset:

Produce ok: <your-topic>-0@15