當(dāng)前位置：首頁(yè) > 编程资源 > 编程问答 >内容正文

编程问答

基于kafka_2.11-2.1.0实现的生产者和消费者代码样例

發(fā)布時(shí)間：2025/7/14 编程问答 22 豆豆

生活随笔收集整理的這篇文章主要介紹了基于kafka_2.11-2.1.0实现的生产者和消费者代码样例小編覺(jué)得挺不錯(cuò)的,現(xiàn)在分享給大家,幫大家做個(gè)參考.

1、搭建部署好zookeeper集群和kafka集群，這里省略。

1 啟動(dòng)zk： 2 bin/zkServer.sh start conf/zoo.cfg。 3 驗(yàn)證zk是否啟動(dòng)成功： 4 bin/zkServer.sh status conf/zoo.cfg。 5 啟動(dòng)kafka： 6 bin/kafka-server-start.sh -daemon config/server.properties。

2、生產(chǎn)者和消費(fèi)者代碼如下所示：

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 // 構(gòu)造一個(gè)java.util.Properties對(duì)象 21 Properties props = new Properties(); 22 // 指定bootstrap.servers屬性。必填，無(wú)默認(rèn)值。用于創(chuàng)建向kafka broker服務(wù)器的連接。 23 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092"); 24 // 指定key.serializer屬性。必填，無(wú)默認(rèn)值。被發(fā)送到broker端的任何消息的格式都必須是字節(jié)數(shù)組。 25 // 因此消息的各個(gè)組件都必須首先做序列化，然后才能發(fā)送到broker。該參數(shù)就是為消息的key做序列化只用的。 26 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 27 // 指定value.serializer屬性。必填，無(wú)默認(rèn)值。和key.serializer類(lèi)似。此被用來(lái)對(duì)消息體即消息value部分做序列化。 28 // 將消息value部分轉(zhuǎn)換成字節(jié)數(shù)組。 29 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 30 //acks參數(shù)用于控制producer生產(chǎn)消息的持久性（durability）。參數(shù)可選值，0、1、-1（all）。 31 props.put("acks", "-1"); 32 //props.put(ProducerConfig.ACKS_CONFIG, "1"); 33 //在producer內(nèi)部自動(dòng)實(shí)現(xiàn)了消息重新發(fā)送。默認(rèn)值0代表不進(jìn)行重試。 34 props.put("retries", 3); 35 //props.put(ProducerConfig.RETRIES_CONFIG, 3); 36 //調(diào)優(yōu)producer吞吐量和延時(shí)性能指標(biāo)都有非常重要作用。默認(rèn)值16384即16KB。 37 props.put("batch.size", 323840); 38 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840); 39 //控制消息發(fā)送延時(shí)行為的，該參數(shù)默認(rèn)值是0。表示消息需要被立即發(fā)送，無(wú)須關(guān)系batch是否被填滿。 40 props.put("linger.ms", 10); 41 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10); 42 //指定了producer端用于緩存消息的緩沖區(qū)的大小，單位是字節(jié)，默認(rèn)值是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 //設(shè)置producer段是否壓縮消息，默認(rèn)值是none。即不壓縮消息。GZIP、Snappy、LZ4 48 //props.put("compression.type", "none"); 49 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none"); 50 //該參數(shù)用于控制producer發(fā)送請(qǐng)求的大小。producer端能夠發(fā)送的最大消息大小。 51 //props.put("max.request.size", 10485760); 52 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760); 53 //producer發(fā)送請(qǐng)求給broker后，broker需要在規(guī)定時(shí)間范圍內(nèi)將處理結(jié)果返還給producer。默認(rèn)30s 54 //props.put("request.timeout.ms", 60000); 55 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000); 56 57 58 // 使用上面創(chuàng)建的Properties對(duì)象構(gòu)造KafkaProducer對(duì)象 59 //如果采用這種方式創(chuàng)建producer，那么就不需要顯示的在Properties中指定key和value序列化類(lèi)了呢。 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 //構(gòu)造好kafkaProducer實(shí)例以后，下一步就是構(gòu)造消息實(shí)例。 67 producer.send(new ProducerRecord<>("topic1", Integer.toString(i), Integer.toString(i))); 68 // 構(gòu)造待發(fā)送的消息對(duì)象ProduceRecord的對(duì)象，指定消息要發(fā)送到的topic主題，分區(qū)以及對(duì)應(yīng)的key和value鍵值對(duì)。 69 // 注意，分區(qū)和key信息可以不用指定，由kafka自行確定目標(biāo)分區(qū)。 70 //ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("my-topic", 71 // Integer.toString(i), Integer.toString(i)); 72 // 調(diào)用kafkaProduce的send方法發(fā)送消息 73 //producer.send(producerRecord); 74 } 75 System.out.println("消息生產(chǎn)結(jié)束......"); 76 // 關(guān)閉kafkaProduce對(duì)象 77 producer.close(); 78 System.out.println("關(guān)閉生產(chǎn)者......"); 79 } 80 81 }

消費(fèi)者代碼如下所示：

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 //構(gòu)造java.util.Properties對(duì)象 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 // 從最早的消息開(kāi)始讀取 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 // 使用創(chuàng)建的Properties實(shí)例構(gòu)造consumer實(shí)例 38 KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props); 39 // 訂閱topic。調(diào)用kafkaConsumer.subscribe方法訂閱consumer group所需的topic列表 40 consumer.subscribe(Arrays.asList(topicName)); 41 try { 42 while (true) { 43 //循環(huán)調(diào)用kafkaConsumer.poll方法獲取封裝在ConsumerRecord的topic消息。 44 ConsumerRecords<String, String> records = consumer.poll(1000); 45 //獲取到封裝在ConsumerRecords消息以后，處理獲取到ConsumerRecord對(duì)象。 46 for (ConsumerRecord<String, String> record : records) { 47 //簡(jiǎn)單的打印輸出 48 System.out.println( 49 "offset = " + record.offset() 50 + ",key = " + record.key() 51 + ",value =" + record.value()); 52 } 53 } 54 } catch (Exception e) { 55 //關(guān)閉kafkaConsumer 56 System.out.println("消息消費(fèi)結(jié)束......"); 57 consumer.close(); 58 } 59 System.out.println("關(guān)閉消費(fèi)者......"); 60 } 61 }

遇到的坑，一開(kāi)始報(bào)的錯(cuò)誤莫名其妙，一開(kāi)始以為使用的jar包版本問(wèn)題，又是報(bào)slf4j的錯(cuò)誤，又是報(bào)log4j的錯(cuò)誤，又是報(bào)空指針的異常。最后百度意外遇到了可能是本地沒(méi)有將ip地址放到hosts文件里面，果然是這個(gè)問(wèn)題。

添加如下所示即可：

然后就可以開(kāi)心的生產(chǎn)消息和消費(fèi)消息了啊。開(kāi)心。

項(xiàng)目結(jié)構(gòu)如下所示：

3、生產(chǎn)者生產(chǎn)消息異步或者同步發(fā)送消息的案例使用：

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 * 同步發(fā)送 19 * Synchronization 同步 20 */ 21 public class ProducerSynchronization { 22 23 public static void main(String[] args) { 24 // 構(gòu)造一個(gè)java.util.Properties對(duì)象 25 Properties props = new Properties(); 26 // 指定bootstrap.servers屬性。必填，無(wú)默認(rèn)值。用于創(chuàng)建向kafka broker服務(wù)器的連接。 27 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092"); 28 // 指定key.serializer屬性。必填，無(wú)默認(rèn)值。被發(fā)送到broker端的任何消息的格式都必須是字節(jié)數(shù)組。 29 // 因此消息的各個(gè)組件都必須首先做序列化，然后才能發(fā)送到broker。該參數(shù)就是為消息的key做序列化只用的。 30 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 31 // 指定value.serializer屬性。必填，無(wú)默認(rèn)值。和key.serializer類(lèi)似。此被用來(lái)對(duì)消息體即消息value部分做序列化。 32 // 將消息value部分轉(zhuǎn)換成字節(jié)數(shù)組。 33 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 34 //acks參數(shù)用于控制producer生產(chǎn)消息的持久性（durability）。參數(shù)可選值，0、1、-1（all）。 35 props.put("acks", "-1"); 36 //props.put(ProducerConfig.ACKS_CONFIG, "1"); 37 //在producer內(nèi)部自動(dòng)實(shí)現(xiàn)了消息重新發(fā)送。默認(rèn)值0代表不進(jìn)行重試。 38 props.put("retries", 3); 39 //props.put(ProducerConfig.RETRIES_CONFIG, 3); 40 //調(diào)優(yōu)producer吞吐量和延時(shí)性能指標(biāo)都有非常重要作用。默認(rèn)值16384即16KB。 41 props.put("batch.size", 323840); 42 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840); 43 //控制消息發(fā)送延時(shí)行為的，該參數(shù)默認(rèn)值是0。表示消息需要被立即發(fā)送，無(wú)須關(guān)系batch是否被填滿。 44 props.put("linger.ms", 10); 45 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10); 46 //指定了producer端用于緩存消息的緩沖區(qū)的大小，單位是字節(jié)，默認(rèn)值是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 //設(shè)置producer段是否壓縮消息，默認(rèn)值是none。即不壓縮消息。GZIP、Snappy、LZ4 52 //props.put("compression.type", "none"); 53 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none"); 54 //該參數(shù)用于控制producer發(fā)送請(qǐng)求的大小。producer端能夠發(fā)送的最大消息大小。 55 //props.put("max.request.size", 10485760); 56 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760); 57 //producer發(fā)送請(qǐng)求給broker后，broker需要在規(guī)定時(shí)間范圍內(nèi)將處理結(jié)果返還給producer。默認(rèn)30s 58 //props.put("request.timeout.ms", 60000); 59 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000); 60 61 62 // 使用上面創(chuàng)建的Properties對(duì)象構(gòu)造KafkaProducer對(duì)象 63 //如果采用這種方式創(chuàng)建producer，那么就不需要顯示的在Properties中指定key和value序列化類(lèi)了呢。 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 //構(gòu)造好kafkaProducer實(shí)例以后，下一步就是構(gòu)造消息實(shí)例。 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) + " 萬(wàn)年呢！！！")); 75 //同步發(fā)送，調(diào)用get()方法無(wú)限等待返回結(jié)果 76 RecordMetadata recordMetadata = future.get(); 77 //成功返回RecordMetadata實(shí)例（包含發(fā)送的元數(shù)據(jù)信息） 78 System.out.println("第 " + i + " 條, " + recordMetadata.toString()); 79 } catch (InterruptedException e) { 80 e.printStackTrace(); 81 } catch (ExecutionException e) { 82 e.printStackTrace(); 83 } 84 85 // 構(gòu)造待發(fā)送的消息對(duì)象ProduceRecord的對(duì)象，指定消息要發(fā)送到的topic主題，分區(qū)以及對(duì)應(yīng)的key和value鍵值對(duì)。 86 // 注意，分區(qū)和key信息可以不用指定，由kafka自行確定目標(biāo)分區(qū)。 87 //ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("my-topic", 88 // Integer.toString(i), Integer.toString(i)); 89 // 調(diào)用kafkaProduce的send方法發(fā)送消息 90 //producer.send(producerRecord); 91 } 92 System.out.println("消息生產(chǎn)結(jié)束......"); 93 // 關(guān)閉kafkaProduce對(duì)象 94 producer.close(); 95 System.out.println("關(guān)閉生產(chǎn)者......"); 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 * 異常發(fā)送 20 * Asynchronous 異步 21 */ 22 public class ProducerAsynchronous { 23 24 public static void main(String[] args) { 25 // 構(gòu)造一個(gè)java.util.Properties對(duì)象 26 Properties props = new Properties(); 27 // 指定bootstrap.servers屬性。必填，無(wú)默認(rèn)值。用于創(chuàng)建向kafka broker服務(wù)器的連接。 28 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092"); 29 // 指定key.serializer屬性。必填，無(wú)默認(rèn)值。被發(fā)送到broker端的任何消息的格式都必須是字節(jié)數(shù)組。 30 // 因此消息的各個(gè)組件都必須首先做序列化，然后才能發(fā)送到broker。該參數(shù)就是為消息的key做序列化只用的。 31 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 32 // 指定value.serializer屬性。必填，無(wú)默認(rèn)值。和key.serializer類(lèi)似。此被用來(lái)對(duì)消息體即消息value部分做序列化。 33 // 將消息value部分轉(zhuǎn)換成字節(jié)數(shù)組。 34 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 35 //acks參數(shù)用于控制producer生產(chǎn)消息的持久性（durability）。參數(shù)可選值，0、1、-1（all）。 36 props.put("acks", "-1"); 37 //props.put(ProducerConfig.ACKS_CONFIG, "1"); 38 //在producer內(nèi)部自動(dòng)實(shí)現(xiàn)了消息重新發(fā)送。默認(rèn)值0代表不進(jìn)行重試。 39 props.put("retries", 3); 40 //props.put(ProducerConfig.RETRIES_CONFIG, 3); 41 //調(diào)優(yōu)producer吞吐量和延時(shí)性能指標(biāo)都有非常重要作用。默認(rèn)值16384即16KB。 42 props.put("batch.size", 323840); 43 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840); 44 //控制消息發(fā)送延時(shí)行為的，該參數(shù)默認(rèn)值是0。表示消息需要被立即發(fā)送，無(wú)須關(guān)系batch是否被填滿。 45 props.put("linger.ms", 10); 46 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10); 47 //指定了producer端用于緩存消息的緩沖區(qū)的大小，單位是字節(jié)，默認(rèn)值是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 //設(shè)置producer段是否壓縮消息，默認(rèn)值是none。即不壓縮消息。GZIP、Snappy、LZ4 53 //props.put("compression.type", "none"); 54 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none"); 55 //該參數(shù)用于控制producer發(fā)送請(qǐng)求的大小。producer端能夠發(fā)送的最大消息大小。 56 //props.put("max.request.size", 10485760); 57 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760); 58 //producer發(fā)送請(qǐng)求給broker后，broker需要在規(guī)定時(shí)間范圍內(nèi)將處理結(jié)果返還給producer。默認(rèn)30s 59 //props.put("request.timeout.ms", 60000); 60 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000); 61 62 63 // 使用上面創(chuàng)建的Properties對(duì)象構(gòu)造KafkaProducer對(duì)象 64 //如果采用這種方式創(chuàng)建producer，那么就不需要顯示的在Properties中指定key和value序列化類(lèi)了呢。 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 //構(gòu)造好kafkaProducer實(shí)例以后，下一步就是構(gòu)造消息實(shí)例。 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 + " 萬(wàn)年呢！！！")); 75 ProducerRecord<String, String> record = new ProducerRecord<String, String>("topic1", Integer.toString(i), "biehl ? wj "+ i +" 萬(wàn)年呢！！！"); 76 //異步發(fā)送 77 producer.send(record, new Callback() { 78 79 @Override 80 public void onCompletion(RecordMetadata metadata, Exception exception) { 81 if(exception == null) { 82 //exception == null代表消息發(fā)送成功 83 System.out.println("消息發(fā)送成功......"); 84 }else { 85 //消息發(fā)送失敗，執(zhí)行錯(cuò)誤的邏輯 86 System.out.println("消息發(fā)送失敗......"); 87 if(exception instanceof RetriableException) { 88 //處理可重試瞬時(shí)異常 89 //... 90 }else { 91 //處理不可重試異常 92 //... 93 } 94 95 } 96 } 97 }); 98 99 100 // 構(gòu)造待發(fā)送的消息對(duì)象ProduceRecord的對(duì)象，指定消息要發(fā)送到的topic主題，分區(qū)以及對(duì)應(yīng)的key和value鍵值對(duì)。 101 // 注意，分區(qū)和key信息可以不用指定，由kafka自行確定目標(biāo)分區(qū)。 102 //ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("my-topic", 103 // Integer.toString(i), Integer.toString(i)); 104 // 調(diào)用kafkaProduce的send方法發(fā)送消息 105 //producer.send(producerRecord); 106 } 107 System.out.println("消息生產(chǎn)結(jié)束......"); 108 // 關(guān)閉kafkaProduce對(duì)象 109 producer.close(); 110 System.out.println("關(guān)閉生產(chǎn)者......"); 111 } 112 113 }

?4、kafka自定義分區(qū)的使用。可以根據(jù)分區(qū)，將指定的key消息發(fā)送到指定的分區(qū)，或者將value消息發(fā)送到指定的分區(qū)。

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、自定義分區(qū) 可以將指定消息發(fā)送到指定的分區(qū) 17 */ 18 public class LovePartitioner implements Partitioner { 19 20 // 隨機(jī)數(shù) 21 private Random random; 22 23 @Override 24 public void configure(Map<String, ?> configs) { 25 // 該方法實(shí)現(xiàn)必要資源的初始化工作 26 random = new Random(); 27 } 28 29 @Override 30 public void close() { 31 //該方法實(shí)現(xiàn)必要的清理工作 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 // 獲取到集群的元數(shù)據(jù)信息 42 List<PartitionInfo> partitionsForTopic = cluster.availablePartitionsForTopic(topic); 43 //遍歷分區(qū)元數(shù)據(jù)信息 44 Iterator<PartitionInfo> it = partitionsForTopic.iterator(); 45 while(it.hasNext()) { 46 PartitionInfo partitionInfo = it.next(); 47 System.out.println("topic消息信息： " + partitionInfo.topic() + " , partition分區(qū)信息： " + partitionInfo.partition() 48 + " ,leader信息： " + partitionInfo.leader() + " , replicas備份信息： " + partitionInfo.replicas()); 49 } 50 // 獲取到分區(qū)的數(shù)量 51 int partitionCount = partitionsForTopic.size(); 52 // 獲取到最后一個(gè)分區(qū) 53 int lovePartition = partitionCount - 1; 54 //如果key不為空且不是love消息，就將隨機(jī)分發(fā)到除最后一個(gè)分區(qū)的其他分區(qū)，否則，分發(fā)到最后一個(gè)分區(qū)。 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 * 異常發(fā)送 18 * Asynchronous 異步 19 */ 20 public class ProducerPartitions { 21 22 public static void main(String[] args) { 23 // 構(gòu)造一個(gè)java.util.Properties對(duì)象 24 Properties props = new Properties(); 25 // 指定bootstrap.servers屬性。必填，無(wú)默認(rèn)值。用于創(chuàng)建向kafka broker服務(wù)器的連接。 26 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092"); 27 // 指定key.serializer屬性。必填，無(wú)默認(rèn)值。被發(fā)送到broker端的任何消息的格式都必須是字節(jié)數(shù)組。 28 // 因此消息的各個(gè)組件都必須首先做序列化，然后才能發(fā)送到broker。該參數(shù)就是為消息的key做序列化只用的。 29 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 30 // 指定value.serializer屬性。必填，無(wú)默認(rèn)值。和key.serializer類(lèi)似。此被用來(lái)對(duì)消息體即消息value部分做序列化。 31 // 將消息value部分轉(zhuǎn)換成字節(jié)數(shù)組。 32 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 33 //acks參數(shù)用于控制producer生產(chǎn)消息的持久性（durability）。參數(shù)可選值，0、1、-1（all）。 34 props.put("acks", "-1"); 35 //props.put(ProducerConfig.ACKS_CONFIG, "1"); 36 //在producer內(nèi)部自動(dòng)實(shí)現(xiàn)了消息重新發(fā)送。默認(rèn)值0代表不進(jìn)行重試。 37 props.put("retries", 3); 38 //props.put(ProducerConfig.RETRIES_CONFIG, 3); 39 //調(diào)優(yōu)producer吞吐量和延時(shí)性能指標(biāo)都有非常重要作用。默認(rèn)值16384即16KB。 40 props.put("batch.size", 323840); 41 //props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840); 42 //控制消息發(fā)送延時(shí)行為的，該參數(shù)默認(rèn)值是0。表示消息需要被立即發(fā)送，無(wú)須關(guān)系batch是否被填滿。 43 props.put("linger.ms", 10); 44 //props.put(ProducerConfig.LINGER_MS_CONFIG, 10); 45 //指定了producer端用于緩存消息的緩沖區(qū)的大小，單位是字節(jié)，默認(rèn)值是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 //設(shè)置producer段是否壓縮消息，默認(rèn)值是none。即不壓縮消息。GZIP、Snappy、LZ4 51 //props.put("compression.type", "none"); 52 //props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none"); 53 //該參數(shù)用于控制producer發(fā)送請(qǐng)求的大小。producer端能夠發(fā)送的最大消息大小。 54 //props.put("max.request.size", 10485760); 55 //props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760); 56 //producer發(fā)送請(qǐng)求給broker后，broker需要在規(guī)定時(shí)間范圍內(nèi)將處理結(jié)果返還給producer。默認(rèn)30s 57 //props.put("request.timeout.ms", 60000); 58 //props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000); 59 60 //指定自定義分區(qū) 61 props.put("partitioner.class", "com.bie.kafka.partitioner.LovePartitioner"); 62 //props.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "com.bie.kafka.partitioner.LovePartitioner"); 63 64 // 使用上面創(chuàng)建的Properties對(duì)象構(gòu)造KafkaProducer對(duì)象 65 //如果采用這種方式創(chuàng)建producer，那么就不需要顯示的在Properties中指定key和value序列化類(lèi)了呢。 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 //無(wú)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 //生產(chǎn)消息 81 try { 82 //發(fā)送無(wú)key消息 83 RecordMetadata recordMetadata1 = producer.send(nonKeyRecord).get(); 84 System.out.println(recordMetadata1.toString()); 85 86 //發(fā)送key不為love的消息 87 RecordMetadata recordMetadata2 = producer.send(nonLoveRecord).get(); 88 System.out.println(recordMetadata2.toString()); 89 90 91 //發(fā)送key為love的消息 92 RecordMetadata recordMetadata3 = producer.send(loveRecord).get(); 93 System.out.println(recordMetadata3.toString()); 94 95 //發(fā)送無(wú)key消息 96 RecordMetadata recordMetadata11 = producer.send(nonKeyRecord).get(); 97 System.out.println(recordMetadata11.toString()); 98 99 //發(fā)送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("消息生產(chǎn)結(jié)束......"); 109 // 關(guān)閉kafkaProduce對(duì)象 110 producer.close(); 111 System.out.println("關(guān)閉生產(chǎn)者......"); 112 } 113 114 }

執(zhí)行完生產(chǎn)者生產(chǎn)消息以后，可以使用命令進(jìn)行查看：

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("自定義序列化失敗告警.....請(qǐng)緊急處理", 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 異常發(fā)送 Asynchronous 異步 17 */ 18 public class ProducerPartitions { 19 20 public static void main(String[] args) { 21 // 構(gòu)造一個(gè)java.util.Properties對(duì)象 22 Properties props = new Properties(); 23 // 指定bootstrap.servers屬性。必填，無(wú)默認(rèn)值。用于創(chuàng)建向kafka broker服務(wù)器的連接。 24 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092"); 25 // 指定key.serializer屬性。必填，無(wú)默認(rèn)值。被發(fā)送到broker端的任何消息的格式都必須是字節(jié)數(shù)組。 26 // 因此消息的各個(gè)組件都必須首先做序列化，然后才能發(fā)送到broker。該參數(shù)就是為消息的key做序列化只用的。 27 // props.put("key.serializer", 28 // "org.apache.kafka.common.serialization.StringSerializer"); 29 // 指定value.serializer屬性。必填，無(wú)默認(rèn)值。和key.serializer類(lèi)似。此被用來(lái)對(duì)消息體即消息value部分做序列化。 30 // 將消息value部分轉(zhuǎn)換成字節(jié)數(shù)組。 31 // props.put("value.serializer", 32 // "org.apache.kafka.common.serialization.StringSerializer"); 33 // acks參數(shù)用于控制producer生產(chǎn)消息的持久性（durability）。參數(shù)可選值，0、1、-1（all）。 34 props.put("acks", "-1"); 35 // props.put(ProducerConfig.ACKS_CONFIG, "1"); 36 // 在producer內(nèi)部自動(dòng)實(shí)現(xiàn)了消息重新發(fā)送。默認(rèn)值0代表不進(jìn)行重試。 37 props.put("retries", 3); 38 // props.put(ProducerConfig.RETRIES_CONFIG, 3); 39 // 調(diào)優(yōu)producer吞吐量和延時(shí)性能指標(biāo)都有非常重要作用。默認(rèn)值16384即16KB。 40 props.put("batch.size", 323840); 41 // props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840); 42 // 控制消息發(fā)送延時(shí)行為的，該參數(shù)默認(rèn)值是0。表示消息需要被立即發(fā)送，無(wú)須關(guān)系batch是否被填滿。 43 props.put("linger.ms", 10); 44 // props.put(ProducerConfig.LINGER_MS_CONFIG, 10); 45 // 指定了producer端用于緩存消息的緩沖區(qū)的大小，單位是字節(jié)，默認(rèn)值是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 // 設(shè)置producer段是否壓縮消息，默認(rèn)值是none。即不壓縮消息。GZIP、Snappy、LZ4 51 // props.put("compression.type", "none"); 52 // props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none"); 53 // 該參數(shù)用于控制producer發(fā)送請(qǐng)求的大小。producer端能夠發(fā)送的最大消息大小。 54 // props.put("max.request.size", 10485760); 55 // props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760); 56 // producer發(fā)送請(qǐng)求給broker后，broker需要在規(guī)定時(shí)間范圍內(nèi)將處理結(jié)果返還給producer。默認(rèn)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 // 使用上面創(chuàng)建的Properties對(duì)象構(gòu)造KafkaProducer對(duì)象 66 // 如果采用這種方式創(chuàng)建producer，那么就不需要顯示的在Properties中指定key和value序列化類(lèi)了呢。 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 //指定序列化生產(chǎn)的消息 77 ProducerRecord<String, User> userRecord = new ProducerRecord<String, User>(topic, user); 78 79 // 生產(chǎn)消息 80 try { 81 // 發(fā)送無(wú)key消息 82 RecordMetadata recordMetadata = producer.send(userRecord).get(); 83 System.out.println("topic主題：" + recordMetadata.topic() + ", partition分區(qū)" + recordMetadata.partition()); 84 } catch (InterruptedException e) { 85 e.printStackTrace(); 86 } catch (ExecutionException e) { 87 e.printStackTrace(); 88 } 89 System.out.println("消息生產(chǎn)結(jié)束......"); 90 // 關(guān)閉kafkaProduce對(duì)象 91 producer.close(); 92 System.out.println("關(guān)閉生產(chǎn)者......"); 93 } 94 95 }

?6、kafka生產(chǎn)者攔截器鏈的使用。

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;// 失敗的消息數(shù)量 19 private int successCount = 0;// 成功的消息數(shù)量 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 // 構(gòu)造一個(gè)java.util.Properties對(duì)象 26 Properties props = new Properties(); 27 // 指定bootstrap.servers屬性。必填，無(wú)默認(rèn)值。用于創(chuàng)建向kafka broker服務(wù)器的連接。 28 props.put("bootstrap.servers", "192.168.110.130:9092,192.168.110.131:9092,192.168.110.132:9092"); 29 // 指定key.serializer屬性。必填，無(wú)默認(rèn)值。被發(fā)送到broker端的任何消息的格式都必須是字節(jié)數(shù)組。 30 // 因此消息的各個(gè)組件都必須首先做序列化，然后才能發(fā)送到broker。該參數(shù)就是為消息的key做序列化只用的。 31 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 32 // 指定value.serializer屬性。必填，無(wú)默認(rèn)值。和key.serializer類(lèi)似。此被用來(lái)對(duì)消息體即消息value部分做序列化。 33 // 將消息value部分轉(zhuǎn)換成字節(jié)數(shù)組。 34 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); 35 // acks參數(shù)用于控制producer生產(chǎn)消息的持久性（durability）。參數(shù)可選值，0、1、-1（all）。 36 props.put("acks", "-1"); 37 // props.put(ProducerConfig.ACKS_CONFIG, "1"); 38 // 在producer內(nèi)部自動(dòng)實(shí)現(xiàn)了消息重新發(fā)送。默認(rèn)值0代表不進(jìn)行重試。 39 props.put("retries", 3); 40 // props.put(ProducerConfig.RETRIES_CONFIG, 3); 41 // 調(diào)優(yōu)producer吞吐量和延時(shí)性能指標(biāo)都有非常重要作用。默認(rèn)值16384即16KB。 42 props.put("batch.size", 323840); 43 // props.put(ProducerConfig.BATCH_SIZE_CONFIG, 323840); 44 // 控制消息發(fā)送延時(shí)行為的，該參數(shù)默認(rèn)值是0。表示消息需要被立即發(fā)送，無(wú)須關(guān)系batch是否被填滿。 45 props.put("linger.ms", 10); 46 // props.put(ProducerConfig.LINGER_MS_CONFIG, 10); 47 // 指定了producer端用于緩存消息的緩沖區(qū)的大小，單位是字節(jié)，默認(rèn)值是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 // 設(shè)置producer段是否壓縮消息，默認(rèn)值是none。即不壓縮消息。GZIP、Snappy、LZ4 53 // props.put("compression.type", "none"); 54 // props.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "none"); 55 // 該參數(shù)用于控制producer發(fā)送請(qǐng)求的大小。producer端能夠發(fā)送的最大消息大小。 56 // props.put("max.request.size", 10485760); 57 // props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, 10485760); 58 // producer發(fā)送請(qǐng)求給broker后，broker需要在規(guī)定時(shí)間范圍內(nèi)將處理結(jié)果返還給producer。默認(rèn)30s 59 // props.put("request.timeout.ms", 60000); 60 // props.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, 60000); 61 62 // 構(gòu)造連接器鏈 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 // 使用上面創(chuàng)建的Properties對(duì)象構(gòu)造KafkaProducer對(duì)象 69 // 如果采用這種方式創(chuàng)建producer，那么就不需要顯示的在Properties中指定key和value序列化類(lèi)了呢。 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("消息生產(chǎn)結(jié)束......"); 88 // 關(guān)閉kafkaProduce對(duì)象 89 producer.close(); 90 System.out.println("關(guān)閉生產(chǎn)者......"); 91 } 92 93 }

運(yùn)行生產(chǎn)者可以看到可以統(tǒng)計(jì)正確或者錯(cuò)誤消息的格式，運(yùn)行消費(fèi)者可以看到已經(jīng)將時(shí)間戳攔截器的時(shí)間戳加到了消息頭上面。

待續(xù).....

轉(zhuǎn)載于:https://www.cnblogs.com/biehongli/p/10877600.html

總結(jié)

以上是生活随笔為你收集整理的基于kafka_2.11-2.1.0实现的生产者和消费者代码样例的全部?jī)?nèi)容，希望文章能夠幫你解決所遇到的問(wèn)題。

如果覺(jué)得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。

上一篇： SpringBoot基础篇
下一篇：【算法python实现】 -- 岛屿的个