Kafka使用Java客戶端進(jìn)行訪問(wèn)的示例代碼
本文環(huán)境如下:
操作系統(tǒng):CentOS 6 32位
JDK版本:1.8.0_77 32位
Kafka版本:0.9.0.1(Scala 2.11)
1. maven依賴包
<dependency> <groupId>org.apache.kafka</groupId> <artifactId>kafka-clients</artifactId> <version>0.9.0.1</version> </dependency>
2. 生產(chǎn)者代碼
package com.lnho.example.kafka; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import java.util.Properties; public class KafkaProducerExample { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "master: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<>("topic1", Integer.toString(i), Integer.toString(i))); producer.close(); } }
3. 消費(fèi)者代碼
package com.lnho.example.kafka; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; import java.util.Arrays; import java.util.Properties; public class KafkaConsumerExample { public static void main(String[] args) { Properties props = new Properties(); props.put("bootstrap.servers", "master:9092"); props.put("group.id", "test"); props.put("enable.auto.commit", "true"); props.put("auto.commit.interval.ms", "1000"); props.put("session.timeout.ms", "30000"); props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer"); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("topic1")); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(), record.key(), record.value()); } } }
4. 執(zhí)行程序
lib底下需要有:kafka-clients-0.9.0.1.jar log4j-1.2.17.jar slf4j-api-1.7.6.jar slf4j-log4j12-1.7.6.jar
生產(chǎn)者:
java -classpath kafka-example-1.0-SNAPSHOT.jar:lib/* com.lnho.example.kafka.KafkaProducerExample
消費(fèi)者:
java -classpath kafka-example-1.0-SNAPSHOT.jar:lib/* com.lnho.example.kafka.KafkaConsumerExample
以上就是本文的全部?jī)?nèi)容,希望對(duì)大家的學(xué)習(xí)有所幫助,也希望大家多多支持腳本之家。
相關(guān)文章
java獲取文件的inode標(biāo)識(shí)符的方法
這篇文章主要介紹了java獲取文件的inode標(biāo)識(shí)符,本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2022-04-04Struts2中圖片以base64方式上傳至數(shù)據(jù)庫(kù)
這篇文章主要介紹了Struts2中圖片以base64方式上傳至數(shù)據(jù)庫(kù)的實(shí)現(xiàn)代碼,代碼分為前臺(tái)和后臺(tái)兩段,非常不錯(cuò),具有參考借鑒價(jià)值,需要的朋友可以參考下2016-09-09SpringBoot獲取配置文件的簡(jiǎn)單實(shí)現(xiàn)方法
這篇文章主要給大家介紹了關(guān)于SpringBoot如何獲取配置文件的簡(jiǎn)單實(shí)現(xiàn)方法,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家學(xué)習(xí)或者使用Spring Boot具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2020-05-05Java SpringMVC實(shí)現(xiàn)PC端網(wǎng)頁(yè)微信掃碼支付(完整版)
這篇文章主要介紹了Java SpringMVC實(shí)現(xiàn)PC端網(wǎng)頁(yè)微信掃碼支付(完整版)的相關(guān)資料,非常不錯(cuò)具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2016-11-11Java?LocalTime的常用時(shí)間操作總結(jié)
日常開發(fā)中,?我們會(huì)經(jīng)常遇到時(shí)間的運(yùn)算,?操作,?格式化等,?這篇文章主要為大家詳細(xì)介紹了LocalTime的常用時(shí)間操作,感興趣的小伙伴可以了解一下2023-11-11java 對(duì)文件夾目錄進(jìn)行深度遍歷實(shí)例代碼
這篇文章主要介紹了java 對(duì)文件夾目錄進(jìn)行深度遍歷實(shí)例代碼的相關(guān)資料,需要的朋友可以參考下2017-03-03