向kafka集群發(fā)布記錄的kafka客戶端怎么實現(xiàn)

這篇文章主要介紹“向kafka集群發(fā)布記錄的kafka客戶端怎么實現(xiàn)”，在日常操作中，相信很多人在向kafka集群發(fā)布記錄的kafka客戶端怎么實現(xiàn)問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”向kafka集群發(fā)布記錄的kafka客戶端怎么實現(xiàn)”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！

生產(chǎn)者是線程安全的，而且，多線程共享同一個producer實例通常比多個producer實例更快。

這里是一個簡單的例子，使用producer發(fā)送字符串數(shù)據(jù)，包含key和value。

Properties props = new Properties();
 props.put("bootstrap.servers", "localhost:9092");
 props.put("acks", "all");
 props.put("retries", 0);
 props.put("batch.size", 16384);
 props.put("linger.ms", 1);
 props.put("buffer.memory", 33554432);
 props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
 props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

 Producer<String, String> producer = new KafkaProducer<>(props);
 for(int i = 0; i < 100; i++)
     producer.send(new ProducerRecord<String, String>("my-topic", Integer.toString(i), Integer.toString(i)));

 producer.close();

一個producer由幾部分組成：1、一個buff poll，保存尚未發(fā)送的數(shù)據(jù)；2、一個后臺運行的I/O線程，負責(zé)執(zhí)行數(shù)據(jù)發(fā)送。producer使用完畢后，務(wù)必執(zhí)行close操作，否則將會造成資源的泄漏。

send()方法是異步的。當調(diào)用它時，它將記錄添加到緩沖區(qū)中，并立即返回。這使得producer能夠批量的執(zhí)行數(shù)據(jù)的生產(chǎn)。

acks有3個可能的值，0：客戶端不必等待任何的server響應(yīng)；1：leader of partition將會在把數(shù)據(jù)寫入自己的log之后，響應(yīng)客戶端，而不必等待其他的follower完成同步的操作；all：leader和follower全部完成log寫入操作。服務(wù)器才會響應(yīng)客戶端。相比之下，all最慢但是可靠性更好。

如果請求失敗，生產(chǎn)者可以自動重試，但是我們已經(jīng)設(shè)置retries = 0，那么重試將不會發(fā)生。如果我們開啟了重試，可能會出現(xiàn)重復(fù)記錄的問題。

producer保持每個partition的未發(fā)送數(shù)據(jù)的緩沖區(qū)。這些緩沖的大小由batch.size配置指定。如果增大這個配置，可以一次性執(zhí)行更大的批量操作，但需要更多的內(nèi)存（因為我們通常會有一個緩沖區(qū)為每個partition）。

默認情況下，緩沖區(qū)可以立即發(fā)送，即使在緩沖區(qū)中有額外的未使用的空間。但是如果你想減少請求的數(shù)量，可以設(shè)置linger.ms > 0。producer會等待一段時間（單位是毫秒）之后在進行發(fā)送，以期獲得更大的批量操作。例如，在上面的代碼片段，設(shè)置linger.ms = 1, 可能會有100條記錄被批量發(fā)送。但是，如果在1毫秒的時間內(nèi)，沒有跟多的數(shù)據(jù)到達緩沖區(qū)，那么這1毫秒的等待僅僅是增加了延遲，而沒有達到任何正面的效果。需要注意的是，如果在短時間內(nèi)，大量的數(shù)據(jù)到達緩沖區(qū)，即使 linger.ms = 0 ，仍然會發(fā)生批量操作。

buffer.memory控制提供給producer的緩沖內(nèi)存總量，如果該緩沖區(qū)的寫入速率長時間大于輸出速率，那么這個緩沖區(qū)將耗盡。當緩沖區(qū)耗盡后，額外的發(fā)送調(diào)用將被阻塞。阻塞一段時間之后（max.block.ms ），將會拋出一個TimeoutException。

key.serializer 和 value.serializer負責(zé)把record當中key和value 分別轉(zhuǎn)換為byte數(shù)組，kafka提供了一組簡單的序列化class。

到此，關(guān)于“向kafka集群發(fā)布記錄的kafka客戶端怎么實現(xiàn)”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識，請繼續(xù)關(guān)注億速云網(wǎng)站，小編會繼續(xù)努力為大家?guī)砀鄬嵱玫奈恼拢?/p>