[大数据]3.1 异步发送

1.代码示例

public class ProducerSample {
    private static final String topicName = "steven";

    /**
     *  Producer异步发送示例
     */
    public static void producerSend() {
        //Producer配置
        Properties properties = new Properties();
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092");
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        properties.put(ProducerConfig.RETRIES_CONFIG, "0");
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, "16384");
        properties.put(ProducerConfig.LINGER_MS_CONFIG, "1");
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, "33554432");
        //序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        //Producer的主对象
        Producer<String, String> producer = new KafkaProducer<>(properties);

        //消息对象ProducerRecord
        for (int i = 0; i < 10; i++) {
            ProducerRecord<String, String> record = new ProducerRecord<>(topicName, "key-" + i, "value-" + i);
            producer.send(record);
        }

        //所有打开的通道都需要关闭
        producer.close();
    }

    public static void main(String[] args) {
        producerSend();
    }
}

2.代码运行结果
打开一个cmd终端，在E:\Kafka\kafka_2.12-1.1.0\bin\windows目录下执行kafka-console-consumer.bat --zookeeper localhost:2181 --topic steven，可以看到生产的消息。