這篇文章主要講解了“如何使用KafkaAPI-ProducerAPI”，文中的講解內(nèi)容簡單清晰，易于學習與理解，下面請大家跟著小編的思路慢慢深入，一起來研究和學習“如何使用KafkaAPI-ProducerAPI”吧！
十載專注成都網(wǎng)站制作，成都定制網(wǎng)頁設計，個人網(wǎng)站制作服務，為大家分享網(wǎng)站制作知識、方案，網(wǎng)站設計流程、步驟,成功服務上千家企業(yè)。為您提供網(wǎng)站建設,網(wǎng)站制作,網(wǎng)頁設計及定制高端網(wǎng)站建設服務,專注于成都定制網(wǎng)頁設計,高端網(wǎng)頁制作,對搬家公司等多個領(lǐng)域，擁有豐富的網(wǎng)站營銷經(jīng)驗。
Kafka 的 Producer 發(fā)送消息采用的是異步發(fā)送的方式。在消息發(fā)送的過程中，涉及到了兩個線程——main 線程和 Sender 線程，以及一個線程共享變量——RecordAccumulator。main 線程將消息發(fā)送給 RecordAccumulator， Sender 線程不斷從 RecordAccumulator 中拉取消息發(fā)送到 Kafka broker。
相關(guān)參數(shù)：
batch.size： 只有數(shù)據(jù)積累到 batch.size 之后， sender 才會發(fā)送數(shù)據(jù)。
linger.ms： 如果數(shù)據(jù)遲遲未達到 batch.size， sender 等待 linger.time 之后就會發(fā)送數(shù)據(jù)。
org.apache.kafka kafka-clients 2.7.0
需要用到的類：
KafkaProducer：需要創(chuàng)建一個生產(chǎn)者對象，用來發(fā)送數(shù)據(jù)
ProducerConfig：獲取所需的一系列配置參數(shù)
ProducerRecord：每條數(shù)據(jù)都要封裝成一個 ProducerRecord 對象
重慶分公司
重慶分公司
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class MyProducer {
public static void main(String[] args) {
//生產(chǎn)者配置信息可以從ProducerConfig中取Key
//1.創(chuàng)建kafka生產(chǎn)者的配置信息
Properties properties=new Properties();
//2.指定連接的kafka集群
//properties.put("bootstrap.servers","192.168.1.106:9091");
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.106:9091");
//3.ACK應答級別
//properties.put("acks","all");
properties.put(ProducerConfig.ACKS_CONFIG,"all");
//4.重試次數(shù)
//properties.put("retries",3);
properties.put(ProducerConfig.RETRIES_CONFIG,3);
//5.批次大小 16k
//properties.put("batch.size",16384);
properties.put(ProducerConfig.BATCH_SIZE_CONFIG,16384);
//6.等待時間
//properties.put("linger.ms",1);
properties.put(ProducerConfig.LINGER_MS_CONFIG,1);
//7.RecordAccumulator 緩沖區(qū)大小 32M
properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG,33554432);
//properties.put("buffer.memory",33554432);
//8.Key,Value 的序列化類
//properties.put("key.serializer","org.apache.kafka.common.serialization.StringSerializer");
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
//properties.put("value.serializer","org.apache.kafka.common.serialization.StringSerializer");
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringSerializer");
//9.創(chuàng)建生產(chǎn)者對象
KafkaProducer producer = new KafkaProducer<>(properties);
//10.發(fā)送數(shù)據(jù)
for (int i = 0; i < 10; i++) {
ProducerRecord producerRecord = new ProducerRecord<>("first","atguigu--"+i);
producer.send(producerRecord);
}
//11.關(guān)閉資源
producer.close();
}
} 回調(diào)函數(shù)會在 producer 收到 ack 時調(diào)用，為異步調(diào)用， 該方法有兩個參數(shù)，分別是RecordMetadata 和 Exception，如果 Exception 為 null，說明消息發(fā)送成功，如果Exception 不為 null，說明消息發(fā)送失敗。
注意：消息發(fā)送失敗會自動重試，不需要我們在回調(diào)函數(shù)中手動重試。
import org.apache.kafka.clients.producer.*;
import java.util.Properties;
public class CallBackProducer {
public static void main(String[] args) {
//生產(chǎn)者配置信息可以從ProducerConfig中取Key
Properties properties=new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.106:9091,192.168.1.106:9092,192.168.1.106:9093");
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");
//創(chuàng)建生產(chǎn)者對象
KafkaProducer producer = new KafkaProducer<>(properties);
/*創(chuàng)建topic
/opt/kafka/kafka03/bin/kafka-topics.sh --create --zookeeper 192.168.1.106:2181,192.168.1.106:2182,192.168.1.106:2183 --replication-factor 3 --partitions 2 --topic aaa
* */
//發(fā)送數(shù)據(jù)
for (int i = 0; i < 10; i++) {
ProducerRecord producerRecord = new ProducerRecord<>("bbb","d","bbb-atguigu++"+i);
producer.send(producerRecord, (recordMetadata, e) -> {
if (e==null){
System.out.println("aaa "+recordMetadata.partition()+ "--"+recordMetadata.offset());
}else {
e.printStackTrace();
}
});
}
//11.關(guān)閉資源
producer.close();
}
} 同步發(fā)送的意思就是，一條消息發(fā)送之后，會阻塞當前線程， 直至返回 ack。由于 send 方法返回的是一個 Future 對象，根據(jù) Futrue 對象的特點，我們也可以實現(xiàn)同步發(fā)送的效果，只需在調(diào)用 Future 對象的 get 方發(fā)即可、
//10.發(fā)送數(shù)據(jù)
for (int i = 0; i < 10; i++) {
ProducerRecord producerRecord = new ProducerRecord<>("first","atguigu--"+i);
producer.send(producerRecord).get();
} 默認分區(qū)策略源碼：
org.apache.kafka.clients.producer.internals.DefaultPartitioner
import org.apache.kafka.clients.producer.Partitioner;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.PartitionInfo;
import java.util.List;
import java.util.Map;
public class MyPartitioner implements Partitioner {
@Override
public int partition(String topic, Object key, byte[] keyBytes, Object value, byte[] valueBytes, Cluster cluster) {
/*自定義分區(qū)規(guī)則*/
List partitionInfos = cluster.availablePartitionsForTopic(topic);
Integer integer =partitionInfos.size();
return key.toString().hashCode()%integer;
/*指定分區(qū)*/
/* return 1;*/
}
@Override
public void close() {
}
@Override
public void configure(Map map) {
}
} //配置方法 properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.zhl.kafkademo.partitioner.MyPartitioner");
完整代碼：
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import java.util.Properties;
public class PartitionProducer {
public static void main(String[] args) {
//生產(chǎn)者配置信息可以從ProducerConfig中取Key
Properties properties=new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.1.106:9091,192.168.1.106:9092,192.168.1.106:9093");
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");
//配置分區(qū)器的全類名 partitioner.class
properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,"com.zhl.kafkademo.partitioner.MyPartitioner");
//創(chuàng)建生產(chǎn)者對象
KafkaProducer producer = new KafkaProducer<>(properties);
//發(fā)送數(shù)據(jù)
for (int i = 0; i < 10; i++) {
ProducerRecord producerRecord = new ProducerRecord<>("bbb","d","bbb-atguigu++"+i);
producer.send(producerRecord, (recordMetadata, e) -> {
if (e==null){
System.out.println(recordMetadata.topic()+"--"+ recordMetadata.partition()+ "--"+recordMetadata.offset());
}else {
e.printStackTrace();
}
});
}
//11.關(guān)閉資源
producer.close();
}
} 感謝各位的閱讀，以上就是“如何使用KafkaAPI-ProducerAPI”的內(nèi)容了，經(jīng)過本文的學習后，相信大家對如何使用KafkaAPI-ProducerAPI這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識點的文章，歡迎關(guān)注！
