Maven必须配置
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.4.1</version>
</dependency>
Java实现
package com.qf.kafka;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
public class MyProducer {
// 定义TOPIC名称
private final static String TOPIC_NAME = "my-replicated-topic";
public static void main(String[] args) throws ExecutionException, InterruptedException {
// 1.设置参数
// 使用 Properties对象 存放键值对(参数配置信息)
Properties props = new Properties();
// 键值对中存入 bootstrap_server => kafka节点或集群(用逗号隔开)
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.232.136:9092,192.168.232.136:9093,192.168.232.136:9094");
// 把需要发送的key从字符串序列化为字节数组
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 把需要发送的value从字符串序列化为字节数组
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
// 2.创建生成消息的客户端,传入参数
KafkaProducer<String, String> producer = new KafkaProducer<String, String>(props);
// 3.创建消息
// 未指定发送分区,具体发送的分区计算公式:hash(key)%partitionNum
// key:决定了往分区上发;value:具体要发送的消息内容
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME, "mykeyvalue", "hello kafka");
// 4.发送消息,得到消息发送的元数据并输出
RecordMetadata metaData = producer.send(producerRecord).get();
System.out.println("同步方式发送消息结果:" + "topic-" + metaData.topic() + "|partition-" + metaData.partition() + "|offset-" + metaData.offset());
}
}