1、搭建部署好zookeeper集群和kafka集群，这里省略。
1 启动zk： 2 bin/zkServer.sh start conf/zoo.cfg。 3 验证zk是否启动成功： 4 bin/zkServer.sh status conf/zoo.cfg。 5 启动kafka： 6 bin/kafka-server-start.sh -daemon config/server.properties。
2、生产者和消费者代码如下所示：
1 package com.bie.kafka.producer;
2
3 import java.util.Properties;
4
5 import org.apache.kafka.clients.producer.KafkaProducer;
6 import org.apache.kafka.clients.producer.Producer;
7 //import org.apache.kafka.clients.producer.ProducerConfig;
8 import org.apache.kafka.clients.producer.ProducerRecord;
9
10 /**
11 *
12 * @Description TODO
13 * @author biehl
14 * @Date 2019年4月6日 上午11:27:34
15 *
16 */
17 public class ProducerTest {
18
19 public static void main(String[] args) {
20 // 构造一个java.util.Properties对象
21 Properties props = new Properties();
22 // 指定bootstrap.servers属性。必填，无默认值。用于创建向kafka broker服务器的连接。
23 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092");
24 // 指定key.serializer属性。必填，无默认值。被发送到broker端的任何消息的格式都必须是字节数组。
25 // 因此消息的各个组件都必须首先做序列化，然后才能发送到broker。该参数就是为消息的key做序列化只用的。
26 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
27 // 指定value.serializer属性。必填，无默认值。和key.serializer类似。此被用来对消息体即消息value部分做序列化。
28 // 将消息value部分转换成字节数组。
29 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
30 //acks参数用于控制producer生产消息的持久性（durability）。参数可选值，0、1、-1（all）。
31 props.put("acks", "-1");
32 //props.put(ProducerConfig.ACKS_CONFIG, "1");
33 //在producer内部自动实现了消息重新发送。默认值0代表不进行重试。
34 props.put("retries", 3);
35 //props.put(ProducerConfig.RETRIES_CONFIG, 3);
36 //调优producer吞吐量和延时性能指标都有非常重要作用。默认值16384即16KB。
37 props.put("batch.size", 323840);
38 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840);
39 //控制消息发送延时行为的，该参数默认值是0。表示消息需要被立即发送，无须关系batch是否被填满。
40 props.put("linger.ms", 10);
41 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
42 //指定了producer端用于缓存消息的缓冲区的大小，单位是字节，默认值是33554432即32M。
43 props.put("buffer.memory", 33554432);
44 //props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
45 props.put("max.block.ms", 3000);
46 //props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
47 //设置producer段是否压缩消息，默认值是none。即不压缩消息。GZIP、Snappy、LZ4
48 //props.put("compression.type", "none");
49 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
50 //该参数用于控制producer发送请求的大小。producer端能够发送的最大消息大小。
51 //props.put("max.request.size", 10485760);
52 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
53 //producer发送请求给broker后，broker需要在规定时间范围内将处理结果返还给producer。默认30s
54 //props.put("request.timeout.ms", 60000);
55 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
56
57
58 // 使用上面创建的Properties对象构造KafkaProducer对象
59 //如果采用这种方式创建producer，那么就不需要显示的在Properties中指定key和value序列化类了呢。
60 // Serializer<String> keySerializer = new StringSerializer();
61 // Serializer<String> valueSerializer = new StringSerializer();
62 // Producer<String, String> producer = new KafkaProducer<String, String>(props,
63 // keySerializer, valueSerializer);
64 Producer<String, String> producer = new KafkaProducer<>(props);
65 for (int i = 0; i < 100; i++) {
66 //构造好kafkaProducer实例以后，下一步就是构造消息实例。
67 producer.send(new ProducerRecord<>("topic1", Integer.toString(i), Integer.toString(i)));
68 // 构造待发送的消息对象ProduceRecord的对象，指定消息要发送到的topic主题，分区以及对应的key和value键值对。
69 // 注意，分区和key信息可以不用指定，由kafka自行确定目标分区。
70 //ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("my-topic",
71 // Integer.toString(i), Integer.toString(i));
72 // 调用kafkaProduce的send方法发送消息
73 //producer.send(producerRecord);
74 }
75 System.out.println("消息生产结束......");
76 // 关闭kafkaProduce对象
77 producer.close();
78 System.out.println("关闭生产者......");
79 }
80
81 }
消费者代码如下所示：
1 package com.bie.kafka.consumer;
2
3 import java.util.Arrays;
4 import java.util.Properties;
5
6 import org.apache.kafka.clients.consumer.ConsumerRecord;
7 import org.apache.kafka.clients.consumer.ConsumerRecords;
8 import org.apache.kafka.clients.consumer.KafkaConsumer;
9
10 /**
11 *
12 * @Description TODO
13 * @author biehl
14 * @Date 2019年4月6日 下午8:12:28
15 *
16 */
17 public class ConsumerTest {
18
19 public static void main(String[] args) {
20 String topicName = "topic1";
21 String groupId = "group1";
22 //构造java.util.Properties对象
23 Properties props = new Properties();
24 // 必须指定属性。
25 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092");
26 // 必须指定属性。
27 props.put("group.id", groupId);
28 props.put("enable.auto.commit", "true");
29 props.put("auto.commit.interval.ms", "1000");
30 // 从最早的消息开始读取
31 props.put("auto.offset.reset", "earliest");
32 // 必须指定
33 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
34 // 必须指定
35 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
36
37 // 使用创建的Properties实例构造consumer实例
38 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
39 // 订阅topic。调用kafkaConsumer.subscribe方法订阅consumer group所需的topic列表
40 consumer.subscribe(Arrays.asList(topicName));
41 try {
42 while (true) {
43 //循环调用kafkaConsumer.poll方法获取封装在ConsumerRecord的topic消息。
44 ConsumerRecords<String, String> records = consumer.poll(1000);
45 //获取到封装在ConsumerRecords消息以后，处理获取到ConsumerRecord对象。
46 for (ConsumerRecord<String, String> record : records) {
47 //简单的打印输出
48 System.out.println(
49 "offset = " + record.offset()
50 + ",key = " + record.key()
51 + ",value =" + record.value());
52 }
53 }
54 } catch (Exception e) {
55 //关闭kafkaConsumer
56 System.out.println("消息消费结束......");
57 consumer.close();
58 }
59 System.out.println("关闭消费者......");
60 }
61 }
遇到的坑，一开始报的错误莫名其妙，一开始以为使用的jar包版本问题，又是报slf4j的错误，又是报log4j的错误，又是报空指针的异常。最后百度意外遇到了可能是本地没有将ip地址放到hosts文件里面，果然是这个问题。
添加如下所示即可：
然后就可以开心的生产消息和消费消息了啊。开心。
项目结构如下所示：
3、生产者生产消息异步或者同步发送消息的案例使用：
Synchronization 同步
1 package com.bie.kafka.producer;
2
3 import java.util.Properties;
4 import java.util.concurrent.ExecutionException;
5 import java.util.concurrent.Future;
6
7 import org.apache.kafka.clients.producer.KafkaProducer;
8 import org.apache.kafka.clients.producer.Producer;
9 //import org.apache.kafka.clients.producer.ProducerConfig;
10 import org.apache.kafka.clients.producer.ProducerRecord;
11 import org.apache.kafka.clients.producer.RecordMetadata;
12
13 /**
14 *
15 * @Description TODO
16 * @author biehl
17 * @Date 2019年4月6日 上午11:27:34
18 * 同步发送
19 * Synchronization 同步
20 */
21 public class ProducerSynchronization {
22
23 public static void main(String[] args) {
24 // 构造一个java.util.Properties对象
25 Properties props = new Properties();
26 // 指定bootstrap.servers属性。必填，无默认值。用于创建向kafka broker服务器的连接。
27 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092");
28 // 指定key.serializer属性。必填，无默认值。被发送到broker端的任何消息的格式都必须是字节数组。
29 // 因此消息的各个组件都必须首先做序列化，然后才能发送到broker。该参数就是为消息的key做序列化只用的。
30 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
31 // 指定value.serializer属性。必填，无默认值。和key.serializer类似。此被用来对消息体即消息value部分做序列化。
32 // 将消息value部分转换成字节数组。
33 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
34 //acks参数用于控制producer生产消息的持久性（durability）。参数可选值，0、1、-1（all）。
35 props.put("acks", "-1");
36 //props.put(ProducerConfig.ACKS_CONFIG, "1");
37 //在producer内部自动实现了消息重新发送。默认值0代表不进行重试。
38 props.put("retries", 3);
39 //props.put(ProducerConfig.RETRIES_CONFIG, 3);
40 //调优producer吞吐量和延时性能指标都有非常重要作用。默认值16384即16KB。
41 props.put("batch.size", 323840);
42 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840);
43 //控制消息发送延时行为的，该参数默认值是0。表示消息需要被立即发送，无须关系batch是否被填满。
44 props.put("linger.ms", 10);
45 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
46 //指定了producer端用于缓存消息的缓冲区的大小，单位是字节，默认值是33554432即32M。
47 props.put("buffer.memory", 33554432);
48 //props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
49 props.put("max.block.ms", 3000);
50 //props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
51 //设置producer段是否压缩消息，默认值是none。即不压缩消息。GZIP、Snappy、LZ4
52 //props.put("compression.type", "none");
53 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
54 //该参数用于控制producer发送请求的大小。producer端能够发送的最大消息大小。
55 //props.put("max.request.size", 10485760);
56 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
57 //producer发送请求给broker后，broker需要在规定时间范围内将处理结果返还给producer。默认30s
58 //props.put("request.timeout.ms", 60000);
59 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
60
61
62 // 使用上面创建的Properties对象构造KafkaProducer对象
63 //如果采用这种方式创建producer，那么就不需要显示的在Properties中指定key和value序列化类了呢。
64 // Serializer<String> keySerializer = new StringSerializer();
65 // Serializer<String> valueSerializer = new StringSerializer();
66 // Producer<String, String> producer = new KafkaProducer<String, String>(props,
67 // keySerializer, valueSerializer);
68 Producer<String, String> producer = new KafkaProducer<>(props);
69 for (int i = 0; i < 100; i++) {
70 //构造好kafkaProducer实例以后，下一步就是构造消息实例。
71 // producer.send(new ProducerRecord<>("topic1", Integer.toString(i), Integer.toString(i)));
72
73 try {
74 Future<RecordMetadata> future = producer.send(new ProducerRecord<>("topic1",Integer.toString(i),"biehl ? wj " + Integer.toString(i) + " 万年呢！！！"));
75 //同步发送，调用get()方法无限等待返回结果
76 RecordMetadata recordMetadata = future.get();
77 //成功返回RecordMetadata实例（包含发送的元数据信息）
78 System.out.println("第 " + i + " 条, " + recordMetadata.toString());
79 } catch (InterruptedException e) {
80 e.printStackTrace();
81 } catch (ExecutionException e) {
82 e.printStackTrace();
83 }
84
85 // 构造待发送的消息对象ProduceRecord的对象，指定消息要发送到的topic主题，分区以及对应的key和value键值对。
86 // 注意，分区和key信息可以不用指定，由kafka自行确定目标分区。
87 //ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("my-topic",
88 // Integer.toString(i), Integer.toString(i));
89 // 调用kafkaProduce的send方法发送消息
90 //producer.send(producerRecord);
91 }
92 System.out.println("消息生产结束......");
93 // 关闭kafkaProduce对象
94 producer.close();
95 System.out.println("关闭生产者......");
96 }
97
98 }
Asynchronous 异步：
1 package com.bie.kafka.producer;
2
3 import java.util.Properties;
4 import java.util.concurrent.Future;
5
6 import org.apache.kafka.clients.producer.Callback;
7 import org.apache.kafka.clients.producer.KafkaProducer;
8 import org.apache.kafka.clients.producer.Producer;
9 //import org.apache.kafka.clients.producer.ProducerConfig;
10 import org.apache.kafka.clients.producer.ProducerRecord;
11 import org.apache.kafka.clients.producer.RecordMetadata;
12 import org.apache.kafka.common.errors.RetriableException;
13
14 /**
15 *
16 * @Description TODO
17 * @author biehl
18 * @Date 2019年4月6日 上午11:27:34
19 * 异常发送
20 * Asynchronous 异步
21 */
22 public class ProducerAsynchronous {
23
24 public static void main(String[] args) {
25 // 构造一个java.util.Properties对象
26 Properties props = new Properties();
27 // 指定bootstrap.servers属性。必填，无默认值。用于创建向kafka broker服务器的连接。
28 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092");
29 // 指定key.serializer属性。必填，无默认值。被发送到broker端的任何消息的格式都必须是字节数组。
30 // 因此消息的各个组件都必须首先做序列化，然后才能发送到broker。该参数就是为消息的key做序列化只用的。
31 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
32 // 指定value.serializer属性。必填，无默认值。和key.serializer类似。此被用来对消息体即消息value部分做序列化。
33 // 将消息value部分转换成字节数组。
34 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
35 //acks参数用于控制producer生产消息的持久性（durability）。参数可选值，0、1、-1（all）。
36 props.put("acks", "-1");
37 //props.put(ProducerConfig.ACKS_CONFIG, "1");
38 //在producer内部自动实现了消息重新发送。默认值0代表不进行重试。
39 props.put("retries", 3);
40 //props.put(ProducerConfig.RETRIES_CONFIG, 3);
41 //调优producer吞吐量和延时性能指标都有非常重要作用。默认值16384即16KB。
42 props.put("batch.size", 323840);
43 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840);
44 //控制消息发送延时行为的，该参数默认值是0。表示消息需要被立即发送，无须关系batch是否被填满。
45 props.put("linger.ms", 10);
46 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
47 //指定了producer端用于缓存消息的缓冲区的大小，单位是字节，默认值是33554432即32M。
48 props.put("buffer.memory", 33554432);
49 //props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
50 props.put("max.block.ms", 3000);
51 //props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
52 //设置producer段是否压缩消息，默认值是none。即不压缩消息。GZIP、Snappy、LZ4
53 //props.put("compression.type", "none");
54 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
55 //该参数用于控制producer发送请求的大小。producer端能够发送的最大消息大小。
56 //props.put("max.request.size", 10485760);
57 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
58 //producer发送请求给broker后，broker需要在规定时间范围内将处理结果返还给producer。默认30s
59 //props.put("request.timeout.ms", 60000);
60 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
61
62
63 // 使用上面创建的Properties对象构造KafkaProducer对象
64 //如果采用这种方式创建producer，那么就不需要显示的在Properties中指定key和value序列化类了呢。
65 // Serializer<String> keySerializer = new StringSerializer();
66 // Serializer<String> valueSerializer = new StringSerializer();
67 // Producer<String, String> producer = new KafkaProducer<String, String>(props,
68 // keySerializer, valueSerializer);
69 Producer<String, String> producer = new KafkaProducer<>(props);
70 for (int i = 0; i < 100; i++) {
71 //构造好kafkaProducer实例以后，下一步就是构造消息实例。
72 // producer.send(new ProducerRecord<>("topic1", Integer.toString(i), Integer.toString(i)));
73
74 //Future<RecordMetadata> send = producer.send(new ProducerRecord<>("topic1",Integer.toString(i),"biehl ? wj " + i + " 万年呢！！！"));
75 ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic1", Integer.toString(i), "biehl ? wj "+ i +" 万年呢！！！");
76 //异步发送
77 producer.send(record, new Callback() {
78
79 @Override
80 public void onCompletion(RecordMetadata metadata, Exception exception) {
81 if(exception == null) {
82 //exception == null代表消息发送成功
83 System.out.println("消息发送成功......");
84 }else {
85 //消息发送失败，执行错误的逻辑
86 System.out.println("消息发送失败......");
87 if(exception instanceof RetriableException) {
88 //处理可重试瞬时异常
89 //...
90 }else {
91 //处理不可重试异常
92 //...
93 }
94
95 }
96 }
97 });
98
99
100 // 构造待发送的消息对象ProduceRecord的对象，指定消息要发送到的topic主题，分区以及对应的key和value键值对。
101 // 注意，分区和key信息可以不用指定，由kafka自行确定目标分区。
102 //ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("my-topic",
103 // Integer.toString(i), Integer.toString(i));
104 // 调用kafkaProduce的send方法发送消息
105 //producer.send(producerRecord);
106 }
107 System.out.println("消息生产结束......");
108 // 关闭kafkaProduce对象
109 producer.close();
110 System.out.println("关闭生产者......");
111 }
112
113 }
4、kafka自定义分区的使用。可以根据分区，将指定的key消息发送到指定的分区，或者将value消息发送到指定的分区。
1 package com.bie.kafka.partitioner;
2
3 import java.util.Iterator;
4 import java.util.List;
5 import java.util.Map;
6 import java.util.Random;
7
8 import org.apache.kafka.clients.producer.Partitioner;
9 import org.apache.kafka.common.Cluster;
10 import org.apache.kafka.common.PartitionInfo;
11
12 /**
13 *
14 * @Description TODO
15 * @author biehl
16 * @Date 2019年5月18日 上午10:28:48 1、自定义分区 可以将指定消息发送到指定的分区
17 */
18 public class LovePartitioner implements Partitioner {
19
20 // 随机数
21 private Random random;
22
23 @Override
24 public void configure(Map<String, ?> configs) {
25 // 该方法实现必要资源的初始化工作
26 random = new Random();
27 }
28
29 @Override
30 public void close() {
31 //該方法实现必要的清理工作
32 }
33
34 @Override
35 public int partition(String topic, Object keyObject, byte[] keyBytes, Object value, byte[] valueBytes,
36 Cluster cluster) {
37 // 获取到key
38 String key = (String) keyObject;
39 // 打印输出key信息
40 System.out.println("key : " + key);
41 // 获取到集群的元数据信息
42 List<PartitionInfo> partitionsForTopic = cluster.availablePartitionsForTopic(topic);
43 //遍历分区元数据信息
44 Iterator<PartitionInfo> it = partitionsForTopic.iterator();
45 while(it.hasNext()) {
46 PartitionInfo partitionInfo = it.next();
47 System.out.println("topic消息信息： " + partitionInfo.topic() + " , partition分区信息： " + partitionInfo.partition()
48 + " ,leader信息： " + partitionInfo.leader() + " , replicas备份信息： " + partitionInfo.replicas());
49 }
50 // 获取到分区的数量
51 int partitionCount = partitionsForTopic.size();
52 // 获取到最后一个分区
53 int lovePartition = partitionCount - 1;
54 //如果key不为空且不是love消息，就将随机分发到除最后一个分区的其他分区，否则，分发到最后一个分区。
55 return key == null || key.isEmpty() || !key.contains("love")
56 ? random.nextInt(partitionCount - 1)
57 : lovePartition;
58 }
59
60 }
1 package com.bie.kafka.partitioner;
2
3 import java.util.Properties;
4 import java.util.concurrent.ExecutionException;
5
6 import org.apache.kafka.clients.producer.KafkaProducer;
7 import org.apache.kafka.clients.producer.Producer;
8 //import org.apache.kafka.clients.producer.ProducerConfig;
9 import org.apache.kafka.clients.producer.ProducerRecord;
10 import org.apache.kafka.clients.producer.RecordMetadata;
11
12 /**
13 *
14 * @Description TODO
15 * @author biehl
16 * @Date 2019年4月6日 上午11:27:34
17 * 异常发送
18 * Asynchronous 异步
19 */
20 public class ProducerPartitions {
21
22 public static void main(String[] args) {
23 // 构造一个java.util.Properties对象
24 Properties props = new Properties();
25 // 指定bootstrap.servers属性。必填，无默认值。用于创建向kafka broker服务器的连接。
26 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092");
27 // 指定key.serializer属性。必填，无默认值。被发送到broker端的任何消息的格式都必须是字节数组。
28 // 因此消息的各个组件都必须首先做序列化，然后才能发送到broker。该参数就是为消息的key做序列化只用的。
29 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
30 // 指定value.serializer属性。必填，无默认值。和key.serializer类似。此被用来对消息体即消息value部分做序列化。
31 // 将消息value部分转换成字节数组。
32 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
33 //acks参数用于控制producer生产消息的持久性（durability）。参数可选值，0、1、-1（all）。
34 props.put("acks", "-1");
35 //props.put(ProducerConfig.ACKS_CONFIG, "1");
36 //在producer内部自动实现了消息重新发送。默认值0代表不进行重试。
37 props.put("retries", 3);
38 //props.put(ProducerConfig.RETRIES_CONFIG, 3);
39 //调优producer吞吐量和延时性能指标都有非常重要作用。默认值16384即16KB。
40 props.put("batch.size", 323840);
41 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840);
42 //控制消息发送延时行为的，该参数默认值是0。表示消息需要被立即发送，无须关系batch是否被填满。
43 props.put("linger.ms", 10);
44 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
45 //指定了producer端用于缓存消息的缓冲区的大小，单位是字节，默认值是33554432即32M。
46 props.put("buffer.memory", 33554432);
47 //props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
48 props.put("max.block.ms", 3000);
49 //props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
50 //设置producer段是否压缩消息，默认值是none。即不压缩消息。GZIP、Snappy、LZ4
51 //props.put("compression.type", "none");
52 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
53 //该参数用于控制producer发送请求的大小。producer端能够发送的最大消息大小。
54 //props.put("max.request.size", 10485760);
55 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
56 //producer发送请求给broker后，broker需要在规定时间范围内将处理结果返还给producer。默认30s
57 //props.put("request.timeout.ms", 60000);
58 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
59
60 //指定自定义分区
61 props.put("partitioner.class", "com.bie.kafka.partitioner.LovePartitioner");
62 //props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.bie.kafka.partitioner.LovePartitioner");
63
64 // 使用上面创建的Properties对象构造KafkaProducer对象
65 //如果采用这种方式创建producer，那么就不需要显示的在Properties中指定key和value序列化类了呢。
66 // Serializer<String> keySerializer = new StringSerializer();
67 // Serializer<String> valueSerializer = new StringSerializer();
68 // Producer<String, String> producer = new KafkaProducer<String, String>(props,
69 // keySerializer, valueSerializer);
70
71 String topic = "topic1";
72 Producer<Object, String> producer = new KafkaProducer<Object, String>(props);
73 //无key消息
74 ProducerRecord<Object, String> nonKeyRecord = new ProducerRecord<Object, String>(topic, "non-key record");
75 //love消息
76 ProducerRecord<Object, String> loveRecord = new ProducerRecord<Object, String>(topic, "love","biehl ? wj");
77 //非love消息
78 ProducerRecord<Object, String> nonLoveRecord = new ProducerRecord<Object, String>(topic, "other", "non-love record");
79
80 //生产消息
81 try {
82 //发送无key消息
83 RecordMetadata recordMetadata1 = producer.send(nonKeyRecord).get();
84 System.out.println(recordMetadata1.toString());
85
86 //发送key不为love的消息
87 RecordMetadata recordMetadata2 = producer.send(nonLoveRecord).get();
88 System.out.println(recordMetadata2.toString());
89
90
91 //发送key为love的消息
92 RecordMetadata recordMetadata3 = producer.send(loveRecord).get();
93 System.out.println(recordMetadata3.toString());
94
95 //发送无key消息
96 RecordMetadata recordMetadata11 = producer.send(nonKeyRecord).get();
97 System.out.println(recordMetadata11.toString());
98
99 //发送key不为love的消息
100 RecordMetadata recordMetadata22 = producer.send(nonLoveRecord).get();
101 System.out.println(recordMetadata22.toString());
102
103 } catch (InterruptedException e) {
104 e.printStackTrace();
105 } catch (ExecutionException e) {
106 e.printStackTrace();
107 }
108 System.out.println("消息生产结束......");
109 // 关闭kafkaProduce对象
110 producer.close();
111 System.out.println("关闭生产者......");
112 }
113
114 }
执行完生产者生产消息以后，可以使用命令进行查看：
1 [root@slaver1 kafka_2.11-2.1.0]# bin/kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list slaver1:9092 --topic topic1
5、kafka自定义序列化value消息:
1 package com.bie.kafka.serializer;
2
3 /**
4 *
5 * @Description TODO
6 * @author biehl
7 * @Date 2019年5月18日 上午11:34:32
8 *
9 */
10 public class User {
11
12 private int id;
13 private String name;
14 private String number;
15 private int age;
16
17 public int getId() {
18 return id;
19 }
20
21 public void setId(int id) {
22 this.id = id;
23 }
24
25 public String getName() {
26 return name;
27 }
28
29 public void setName(String name) {
30 this.name = name;
31 }
32
33 public String getNumber() {
34 return number;
35 }
36
37 public void setNumber(String number) {
38 this.number = number;
39 }
40
41 public int getAge() {
42 return age;
43 }
44
45 public void setAge(int age) {
46 this.age = age;
47 }
48
49 @Override
50 public String toString() {
51 return "User [id=" + id + ", name=" + name + ", number=" + number + ", age=" + age + "]";
52 }
53
54 public User(int id, String name, String number, int age) {
55 super();
56 this.id = id;
57 this.name = name;
58 this.number = number;
59 this.age = age;
60 }
61
62 public User() {
63 super();
64 }
65
66 }
1 package com.bie.kafka.serializer;
2
3 import java.io.IOException;
4 import java.util.Map;
5
6 import org.apache.kafka.common.serialization.Serializer;
7 import org.apache.log4j.Logger;
8 import org.codehaus.jackson.map.ObjectMapper;
9
10 /**
11 *
12 * @Description TODO
13 * @author biehl
14 * @Date 2019年5月18日 上午11:36:57 1、自定义序列化操作
15 *
16 */
17 public class UserSerializer implements Serializer<Object> {
18
19 private Logger logger = Logger.getLogger(UserSerializer.class);
20
21 //
22 private ObjectMapper objectMapper;
23
24 @Override
25 public void close() {
26
27 }
28
29 @Override
30 public void configure(Map<String, ?> configs, boolean isKey) {
31 objectMapper = new ObjectMapper();
32 }
33
34 @Override
35 public byte[] serialize(String topic, Object data) {
36 byte[] ret = null;
37 try {
38 ret = objectMapper.writeValueAsString(data).getBytes("utf-8");
39 } catch (IOException e) {
40 e.printStackTrace();
41 logger.warn("自定义序列化失败告警.....请紧急处理", e);
42 }
43 return ret;
44 }
45
46 }
1 package com.bie.kafka.serializer;
2
3 import java.util.Properties;
4 import java.util.concurrent.ExecutionException;
5
6 import org.apache.kafka.clients.producer.KafkaProducer;
7 import org.apache.kafka.clients.producer.Producer;
8 //import org.apache.kafka.clients.producer.ProducerConfig;
9 import org.apache.kafka.clients.producer.ProducerRecord;
10 import org.apache.kafka.clients.producer.RecordMetadata;
11
12 /**
13 *
14 * @Description TODO
15 * @author biehl
16 * @Date 2019年4月6日 上午11:27:34 异常发送 Asynchronous 异步
17 */
18 public class ProducerPartitions {
19
20 public static void main(String[] args) {
21 // 构造一个java.util.Properties对象
22 Properties props = new Properties();
23 // 指定bootstrap.servers属性。必填，无默认值。用于创建向kafka broker服务器的连接。
24 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092");
25 // 指定key.serializer属性。必填，无默认值。被发送到broker端的任何消息的格式都必须是字节数组。
26 // 因此消息的各个组件都必须首先做序列化，然后才能发送到broker。该参数就是为消息的key做序列化只用的。
27 // props.put("key.serializer",
28 // "org.apache.kafka.common.serialization.StringSerializer");
29 // 指定value.serializer属性。必填，无默认值。和key.serializer类似。此被用来对消息体即消息value部分做序列化。
30 // 将消息value部分转换成字节数组。
31 // props.put("value.serializer",
32 // "org.apache.kafka.common.serialization.StringSerializer");
33 // acks参数用于控制producer生产消息的持久性（durability）。参数可选值，0、1、-1（all）。
34 props.put("acks", "-1");
35 // props.put(ProducerConfig.ACKS_CONFIG, "1");
36 // 在producer内部自动实现了消息重新发送。默认值0代表不进行重试。
37 props.put("retries", 3);
38 // props.put(ProducerConfig.RETRIES_CONFIG, 3);
39 // 调优producer吞吐量和延时性能指标都有非常重要作用。默认值16384即16KB。
40 props.put("batch.size", 323840);
41 // props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840);
42 // 控制消息发送延时行为的，该参数默认值是0。表示消息需要被立即发送，无须关系batch是否被填满。
43 props.put("linger.ms", 10);
44 // props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
45 // 指定了producer端用于缓存消息的缓冲区的大小，单位是字节，默认值是33554432即32M。
46 props.put("buffer.memory", 33554432);
47 // props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
48 props.put("max.block.ms", 3000);
49 // props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
50 // 设置producer段是否压缩消息，默认值是none。即不压缩消息。GZIP、Snappy、LZ4
51 // props.put("compression.type", "none");
52 // props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
53 // 该参数用于控制producer发送请求的大小。producer端能够发送的最大消息大小。
54 // props.put("max.request.size", 10485760);
55 // props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
56 // producer发送请求给broker后，broker需要在规定时间范围内将处理结果返还给producer。默认30s
57 // props.put("request.timeout.ms", 60000);
58 // props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
59
60 // 自定义序列化操作
61 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
62 // 自定义序列化value的值
63 props.put("value.serializer", "com.bie.kafka.serializer.UserSerializer");
64
65 // 使用上面创建的Properties对象构造KafkaProducer对象
66 // 如果采用这种方式创建producer，那么就不需要显示的在Properties中指定key和value序列化类了呢。
67 // Serializer<String> keySerializer = new StringSerializer();
68 // Serializer<String> valueSerializer = new StringSerializer();
69 // Producer<String, String> producer = new KafkaProducer<String, String>(props,
70 // keySerializer, valueSerializer);
71
72 User user = new User(1,"张三","41076788898765444",25);
73
74 String topic = "topic1";
75 Producer<String, User> producer = new KafkaProducer<String, User>(props);
76 //指定序列化生产的消息
77 ProducerRecord<String, User> userRecord = new ProducerRecord<String, User>(topic, user);
78
79 // 生产消息
80 try {
81 // 发送无key消息
82 RecordMetadata recordMetadata = producer.send(userRecord).get();
83 System.out.println("topic主题：" + recordMetadata.topic() + ", partition分区" + recordMetadata.partition());
84 } catch (InterruptedException e) {
85 e.printStackTrace();
86 } catch (ExecutionException e) {
87 e.printStackTrace();
88 }
89 System.out.println("消息生产结束......");
90 // 关闭kafkaProduce对象
91 producer.close();
92 System.out.println("关闭生产者......");
93 }
94
95 }
6、kafka生产者拦截器链的使用。
1 package com.bie.kafka.interceptor;
2
3 import java.util.Map;
4
5 import org.apache.kafka.clients.producer.ProducerInterceptor;
6 import org.apache.kafka.clients.producer.ProducerRecord;
7 import org.apache.kafka.clients.producer.RecordMetadata;
8
9 /**
10 *
11 * @Description TODO
12 * @author biehl
13 * @Date 2019年5月18日 下午5:06:37 1、拦截器： 可以将拦截器组成拦截器链。
14 *
15 */
16 public class TimeStampPrependerInterceptor implements ProducerInterceptor<String, String> {
17
18 @Override
19 public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
20 return new ProducerRecord<String, String>(record.topic(), record.partition(), record.timestamp(), record.key(),
21 System.currentTimeMillis() + "," + record.value().toLowerCase());
22 }
23
24 @Override
25 public void configure(Map<String, ?> arg0) {
26
27 }
28
29 @Override
30 public void close() {
31
32 }
33
34 @Override
35 public void onAcknowledgement(RecordMetadata arg0, Exception arg1) {
36
37 }
38
39 }
1 package com.bie.kafka.interceptor;
2
3 import java.util.Map;
4
5 import org.apache.kafka.clients.producer.ProducerInterceptor;
6 import org.apache.kafka.clients.producer.ProducerRecord;
7 import org.apache.kafka.clients.producer.RecordMetadata;
8
9 /**
10 *
11 * @Description TODO
12 * @author biehl
13 * @Date 2019年5月18日 下午5:15:04
14 *
15 */
16 public class CounterInterceptor implements ProducerInterceptor<String, String> {
17
18 private int errorCount = 0;// 失败的消息数量
19 private int successCount = 0;// 成功的消息数量
20
21 @Override
22 public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
23
24 return record;
25 }
26
27 @Override
28 public void configure(Map<String, ?> arg0) {
29
30 }
31
32 @Override
33 public void close() {
34 //
35 System.out.println("success count is : " + successCount);
36 System.out.println("error count is : " + errorCount);
37 }
38
39 @Override
40 public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
41 if (exception == null) {
42 successCount++;
43 } else {
44 errorCount++;
45 }
46 }
47
48 }
1 package com.bie.kafka.interceptor;
2
3 import java.util.ArrayList;
4 import java.util.List;
5 import java.util.Properties;
6 import java.util.concurrent.ExecutionException;
7
8 import org.apache.kafka.clients.producer.KafkaProducer;
9 import org.apache.kafka.clients.producer.Producer;
10 import org.apache.kafka.clients.producer.ProducerConfig;
11 //import org.apache.kafka.clients.producer.ProducerConfig;
12 import org.apache.kafka.clients.producer.ProducerRecord;
13 import org.apache.kafka.clients.producer.RecordMetadata;
14
15 /**
16 *
17 * @Description TODO
18 * @author biehl
19 * @Date 2019年4月6日 上午11:27:34 拦截器链的使用
20 *
21 */
22 public class ProducerInterceptor {
23
24 public static void main(String[] args) {
25 // 构造一个java.util.Properties对象
26 Properties props = new Properties();
27 // 指定bootstrap.servers属性。必填，无默认值。用于创建向kafka broker服务器的连接。
28 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092");
29 // 指定key.serializer属性。必填，无默认值。被发送到broker端的任何消息的格式都必须是字节数组。
30 // 因此消息的各个组件都必须首先做序列化，然后才能发送到broker。该参数就是为消息的key做序列化只用的。
31 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
32 // 指定value.serializer属性。必填，无默认值。和key.serializer类似。此被用来对消息体即消息value部分做序列化。
33 // 将消息value部分转换成字节数组。
34 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
35 // acks参数用于控制producer生产消息的持久性（durability）。参数可选值，0、1、-1（all）。
36 props.put("acks", "-1");
37 // props.put(ProducerConfig.ACKS_CONFIG, "1");
38 // 在producer内部自动实现了消息重新发送。默认值0代表不进行重试。
39 props.put("retries", 3);
40 // props.put(ProducerConfig.RETRIES_CONFIG, 3);
41 // 调优producer吞吐量和延时性能指标都有非常重要作用。默认值16384即16KB。
42 props.put("batch.size", 323840);
43 // props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840);
44 // 控制消息发送延时行为的，该参数默认值是0。表示消息需要被立即发送，无须关系batch是否被填满。
45 props.put("linger.ms", 10);
46 // props.put(ProducerConfig.LINGER_MS_CONFIG, 10);
47 // 指定了producer端用于缓存消息的缓冲区的大小，单位是字节，默认值是33554432即32M。
48 props.put("buffer.memory", 33554432);
49 // props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
50 props.put("max.block.ms", 3000);
51 // props.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 3000);
52 // 设置producer段是否压缩消息，默认值是none。即不压缩消息。GZIP、Snappy、LZ4
53 // props.put("compression.type", "none");
54 // props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none");
55 // 该参数用于控制producer发送请求的大小。producer端能够发送的最大消息大小。
56 // props.put("max.request.size", 10485760);
57 // props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760);
58 // producer发送请求给broker后，broker需要在规定时间范围内将处理结果返还给producer。默认30s
59 // props.put("request.timeout.ms", 60000);
60 // props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000);
61
62 // 构造连接器链
63 List<String> interceptors = new ArrayList<String>();
64 interceptors.add("com.bie.kafka.interceptor.CounterInterceptor");
65 interceptors.add("com.bie.kafka.interceptor.TimeStampPrependerInterceptor");
66 props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, interceptors);
67
68 // 使用上面创建的Properties对象构造KafkaProducer对象
69 // 如果采用这种方式创建producer，那么就不需要显示的在Properties中指定key和value序列化类了呢。
70 // Serializer<String> keySerializer = new StringSerializer();
71 // Serializer<String> valueSerializer = new StringSerializer();
72 // Producer<String, String> producer = new KafkaProducer<String, String>(props,
73 // keySerializer, valueSerializer);
74
75 String topic = "topic1";
76 Producer<String, String> producer = new KafkaProducer<String, String>(props);
77 for (int i = 0; i < 100; i++) {
78 ProducerRecord<String, String> record = new ProducerRecord<String, String>(topic, "biehl ? wj " + i + " 年哟");
79 try {
80 RecordMetadata recordMetadata = producer.send(record).get();
81 System.out.println("topic : " + recordMetadata.topic() + ", partition" + recordMetadata.partition());
82 } catch (InterruptedException | ExecutionException e) {
83 e.printStackTrace();
84 }
85 }
86
87 System.out.println("消息生产结束......");
88 // 关闭kafkaProduce对象
89 producer.close();
90 System.out.println("关闭生产者......");
91 }
92
93 }
运行生产者可以看到可以统计正确或者错误消息的格式，运行消费者可以看到已经将时间戳拦截器的时间戳加到了消息头上面。
待续.....
