1
<依賴項> <groupId> org.apache.kafka </groupId> <Artifactid> kafka-clients </artifactid> </artifactid> <版本> 0.9.0.1 </version>
2。生產者代碼
包com.lnho.example.kafka;導入org.apache.kafka.clients.producer.kafkaproducer;導入org.apache.kafka.clients.producer.producer;導入org.apache.kafka.clients.producer.producercord;導入java.util.properties; public類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（“鍵。 props.put（“ value.serializer”，org.apache.kafka.common.serialization.stringserializer'）;生產者<string，string> producer = new kafkaproducer <>（props）; for（int i = 0; i <100; i ++）producer.Send（new ProducterRecord <>（“ topic1”，integer.tostring（i），integer.tostring（i）））; producer.close（）; }}}3。消費者代碼
包com.lnho.example.kafka;導入org.apache.kafka.clients.consumer.consumerrecord;導入org.apache.kafka.clients.consumer.consumerrecords;導入org.apache.kafka.clients.consumer.kafkaconsumer;導入java.util.arrays;導入java.util.properties; public類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（“鍵。 props.put（“ value.deserializer”，“ org.apache.kafka.common.serialization.stringDeserializer”）; kafkaconsumer <string，string> consumer = new kafkaconsumer <>（props）; commuter.subscribe（arrays.aslist（“ topor1”））; while（true）{commuterRecords <string，string> records = computer.poll（100）; for（consumerRecord <string，字符串> record：record）system.out.printf（“ fortset =％d，key =％s，value =％s/n”，record.offset（），record.key（），record.value（））; }}}}以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持武林網。 ，也希望大家多多支持武林網。