文章目錄

• POM 依賴
• 生產者
• 消費者
• 測試

POM 依賴

版本請同使用的kafka服務端的版本保持一致

<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.4.1</version></dependency>

---

生產者

請小伙伴注意一下注釋，這里就不做多余的解釋啦

package com.artisan.kafka.first;import org.apache.kafka.clients.producer.*; import org.apache.kafka.common.serialization.StringSerializer;import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.Future;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 19:45* @mark: show me the code , change the world*/ public class ProduceClient {private static final String TOPIC = "artisanTopic";public static void main(String[] args) throws ExecutionException, InterruptedException {// 屬性設置 Properties properties = new Properties();properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.126.140:9092");properties.put(ProducerConfig.ACKS_CONFIG,"1");properties.put(ProducerConfig.RETRIES_CONFIG,3);properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());// 根據屬性實例化KafkaProducerProducer<String,String> producer = new KafkaProducer<String, String>(properties);// 創建消息 三個參數，分別是 Topic ，消息的 key ，消息的 message 。String message = "mockValue";ProducerRecord<String ,String> producerRecord = new ProducerRecord<>(TOPIC, "mockKey", message);// 發送消息 (同步)Future<RecordMetadata> result = producer.send(producerRecord);// 獲取同步發送的結果RecordMetadata recordMetadata = result.get();System.out.println(String.format("Message[ %s ] sent to Topic: %s || Partition: %s || Offset: %s",message, recordMetadata.topic(), recordMetadata.partition(), recordMetadata.offset()));}}

---

消費者

請小伙伴注意一下注釋，這里就不做多余的解釋啦

package com.artisan.kafka.first;import org.apache.kafka.clients.consumer.*; import org.apache.kafka.common.serialization.StringDeserializer; import org.apache.kafka.common.serialization.StringSerializer;import java.time.Duration; import java.util.Collections; import java.util.Properties;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 20:09* @mark: show me the code , change the world*/ public class ConsumerClient {private static final String TOPIC = "artisanTopic";public static void main(String[] args) {// 屬性設置Properties properties = new Properties();properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG , "192.168.126.140:9092"); // Broker 的地址properties.put(ConsumerConfig.GROUP_ID_CONFIG,"artisan-consumer-group");// 消費者分組properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest"); // 設置消費者分組最初的消費進度為 earliestproperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,"true"); // 是否自動提交消費進度properties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG,1000); // 自動提交消費進度頻率properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName()); // 消息的 key 的反序列化方式properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,StringDeserializer.class.getName());// 消息的 value 的反序列化方式// 根據設置實例化KafkaConsumerConsumer<String,String> consumer = new KafkaConsumer<>(properties);// 訂閱消息consumer.subscribe(Collections.singleton(TOPIC));// 循環拉取消息while (true){// 拉取消息ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(10));// 遍歷處理消息records.forEach(record -> System.out.println(String.format("接收到消息:Key %s || 內容 %s" , record.key(),record.value())));}} }

屬性的話，需要結合kafka的特性來講解，后面的單獨介紹

---

測試

運行Produce

運行消費端

總結

以上是生活随笔為你收集整理的Apache Kafka-生产消费基础篇的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。