---

前言

Kafka的生产者是向Kafka集群发送消息的部分，有很强的自主性，可以自主定义何时发送消息，发送怎样的消息。

但是发送消息无法发送集合列表，只能以key，value的形式发送消息，在设置消息发送的模式时可以参考value值为json格式的内容，可能存储更大量的参数值。

---

一、生产者客户端

（一）实现过程

1、配置生产者参数
2、创建生产者实例
3、生成消息（主题，key，value）
4、发送消息
5、关闭生产者实例

（二）代码实现

import com.lmz.util.KafkaPara;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;

/**
 * Kafka生产者客户端
 * @author zeyue
 */
public class KafkaProducerTest {

    /**
     * 订阅主题名称
     */
    public static final String topic = "test";

    /**
     * 生产者参数配置方法
     * @return 返回配置好生产者参数的Properties类
     */
    public static Properties initConfig() {
        Properties props = new Properties();
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("bootstrap.servers", KafkaPara.brokerList);
        props.put("client.id", "consumer.client.id.demo");

        return props;
    }

    public static void main(String[] args) {
        Properties properties = initConfig();
        //创建Kafka生产者类
        KafkaProducer<String , String> producer = new KafkaProducer<>(properties);


        for (int i = 0 ; i < 20000 ; i++) {
            ProducerRecord<String ,String > producerRecord = new ProducerRecord<>(topic,"test","Hello Kafka" + i);
            producer.send(producerRecord);
        }


        producer.close();

    }

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47