【Java客戶端訪問Kafka】

我是🌟廖志偉🌟，一名🌕Java開發工程師🌕、📝Java領域優質創作者📝、🎉CSDN博客專家🎉、🌹幕后大佬社區創始人🌹，擁有多年一線研發經驗，研究過各種常見框架及中間件的底層原始碼，對于大型分布式、微服務、三高架構（高性能、高并發、高可用）有過實踐架構經驗，

🍊博主：java_wxid
🍊博主：Java廖志偉
🍊社區：幕后大佬

Java客戶端訪問Kafka

引入maven依賴

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

訊息發送端代碼

package com.sky.kafka.kafkaDemo;

import com.alibaba.fastjson.JSON;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class MsgProducer {

    private final static String TOPIC_NAME = "my-replicated-topic";

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.65.60:9092,192.168.65.60:9093,192.168.65.60:9094");
         /*
         發出訊息持久化機制引數
        （1）acks=0： 表示producer不需要等待任何broker確認收到訊息的回復，就可以繼續發送下一條訊息，性能最高，但是最容易丟訊息，
        （2）acks=1： 至少要等待leader已經成功將資料寫入本地log，但是不需要等待所有follower是否成功寫入，就可以繼續發送下一
             條訊息，這種情況下，如果follower沒有成功備份資料，而此時leader又掛掉，則訊息會丟失，
        （3）acks=-1或all： 需要等待 min.insync.replicas(默認為1，推薦配置大于等于2) 這個引數配置的副本個數都成功寫入日志，這種策略會保證
            只要有一個備份存活就不會丟失資料，這是最強的資料保證，一般除非是金融級別，或跟錢打交道的場景才會使用這種配置，
         */
        /*props.put(ProducerConfig.ACKS_CONFIG, "1");
         *//*
        發送失敗會重試，默認重試間隔100ms，重試能保證訊息發送的可靠性，但是也可能造成訊息重復發送，比如網路抖動，所以需要在
        接收者那邊做好訊息接收的冪等性處理
        *//*
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        //重試間隔設定
        props.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
        //設定發送訊息的本地緩沖區，如果設定了該緩沖區，訊息會先發送到本地緩沖區，可以提高訊息發送性能，默認值是33554432，即32MB
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        *//*
        kafka本地執行緒會從緩沖區取資料，批量發送到broker，
        設定批量發送訊息的大小，默認值是16384，即16kb，就是說一個batch滿了16kb就發送出去
        *//*
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        *//*
        默認值是0，意思就是訊息必須立即被發送，但這樣會影響性能
        一般設定10毫秒左右，就是說這個訊息發送完后會進入本地的一個batch，如果10毫秒內，這個batch滿了16kb就會隨batch一起被發送出去
        如果10毫秒內，batch沒滿，那么也必須把訊息發送出去，不能讓訊息的發送延遲時間太長
        *//*
        props.put(ProducerConfig.LINGER_MS_CONFIG, 10);*/
        //把發送的key從字串序列化為位元組陣列
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        //把發送訊息value從字串序列化為位元組陣列
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        Producer<String, String> producer = new KafkaProducer<String, String>(props);

        int msgNum = 5;
        final CountDownLatch countDownLatch = new CountDownLatch(msgNum);
        for (int i = 1; i <= msgNum; i++) {
            Order order = new Order(i, 100 + i, 1, 1000.00);
            //指定發送磁區
            /*ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , 0, order.getOrderId().toString(), JSON.toJSONString(order));*/
            //未指定發送磁區，具體發送的磁區計算公式：hash(key)%partitionNum
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(TOPIC_NAME
                    , order.getOrderId().toString(), JSON.toJSONString(order));

            //等待訊息發送成功的同步阻塞方法
            /*RecordMetadata metadata = producer.send(producerRecord).get();
            System.out.println("同步方式發送訊息結果：" + "topic-" + metadata.topic() + "|partition-"
                    + metadata.partition() + "|offset-" + metadata.offset());*/

            //異步回呼方式發送訊息
            producer.send(producerRecord, new Callback() {
                public void onCompletion(RecordMetadata metadata, Exception exception) {
                    if (exception != null) {
                        System.err.println("發送訊息失敗：" + exception.getStackTrace());

                    }
                    if (metadata != null) {
                        System.out.println("異步方式發送訊息結果：" + "topic-" + metadata.topic() + "|partition-"
                                + metadata.partition() + "|offset-" + metadata.offset());
                    }
                    countDownLatch.countDown();
                }
            });

            //送積分 TODO

        }

        countDownLatch.await(5, TimeUnit.SECONDS);
        producer.close();
    }
}

訊息接收端代碼

package com.sky.kafka.kafkaDemo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class MsgConsumer {
    private final static String TOPIC_NAME = "my-replicated-topic";
    private final static String CONSUMER_GROUP_NAME = "testGroup";

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "106.14.132.94:9092,106.14.132.94:9093,106.14.132.94:9094");
        // 消費分組名
        props.put(ConsumerConfig.GROUP_ID_CONFIG, CONSUMER_GROUP_NAME);
        // 是否自動提交offset，默認就是true
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        // 自動提交offset的間隔時間
        props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
        //props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
        /*
        當消費主題的是一個新的消費組，或者指定offset的消費方式，offset不存在，那么應該如何消費
        latest(默認) ：只消費自己啟動之后發送到主題的訊息
        earliest：第一次從頭開始消費，以后按照消費offset記錄繼續消費，這個需要區別于consumer.seekToBeginning(每次都從頭開始消費)
        */
        //props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
      /*
      consumer給broker發送心跳的間隔時間，broker接收到心跳如果此時有rebalance發生會通過心跳回應將
      rebalance方案下發給consumer，這個時間可以稍微短一點
      */
        props.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 1000);
        /*
        服務端broker多久感知不到一個consumer心跳就認為他故障了，會將其踢出消費組，
        對應的Partition也會被重新分配給其他consumer，默認是10秒
        */
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 10 * 1000);
        //一次poll最大拉取訊息的條數，如果消費者處理速度很快，可以設定大點，如果處理速度一般，可以設定小點
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500);
        /*
        如果兩次poll操作間隔超過了這個時間，broker就會認為這個consumer處理能力太弱，
        會將其踢出消費組，將磁區分配給別的consumer消費
        */
        props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 30 * 1000);
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);

        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        // 消費指定磁區
        //consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));

        //訊息回溯消費
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seekToBeginning(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));*/

        //指定offset消費
        /*consumer.assign(Arrays.asList(new TopicPartition(TOPIC_NAME, 0)));
        consumer.seek(new TopicPartition(TOPIC_NAME, 0), 10);*/

        //從指定時間點開始消費
        /*List<PartitionInfo> topicPartitions = consumer.partitionsFor(TOPIC_NAME);
        //從1小時前開始消費
        long fetchDataTime = new Date().getTime() - 1000 * 60 * 60;
        Map<TopicPartition, Long> map = new HashMap<>();
        for (PartitionInfo par : topicPartitions) {
            map.put(new TopicPartition(topicName, par.partition()), fetchDataTime);
        }
        Map<TopicPartition, OffsetAndTimestamp> parMap = consumer.offsetsForTimes(map);
        for (Map.Entry<TopicPartition, OffsetAndTimestamp> entry : parMap.entrySet()) {
            TopicPartition key = entry.getKey();
            OffsetAndTimestamp value = entry.getValue();
            if (key == null || value == null) continue;
            Long offset = value.offset();
            System.out.println("partition-" + key.partition() + "|offset-" + offset);
            System.out.println();
            //根據消費里的timestamp確定offset
            if (value != null) {
                consumer.assign(Arrays.asList(key));
                consumer.seek(key, offset);
            }
        }*/

        while (true) {
            /*
             * poll() API 是拉取訊息的長輪詢
             */
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(1000));
            for (ConsumerRecord<String, String> record : records) {
                System.out.printf("收到訊息：partition = %d,offset = %d, key = %s, value = %s%n", record.partition(),
                        record.offset(), record.key(), record.value());
            }

            /*if (records.count() > 0) {
                // 手動同步提交offset，當前執行緒會阻塞直到offset提交成功
                // 一般使用同步提交，因為提交之后一般也沒有什么邏輯代碼了
                consumer.commitSync();

                // 手動異步提交offset，當前執行緒提交offset不會阻塞，可以繼續處理后面的程式邏輯
                consumer.commitAsync(new OffsetCommitCallback() {
                    @Override
                    public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
                        if (exception != null) {
                            System.err.println("Commit failed for " + offsets);
                            System.err.println("Commit failed exception: " + exception.getStackTrace());
                        }
                    }
                });

            }*/
        }
    }
}

總結

以上就是今天要講的內容，還希望各位讀者大大能夠在評論區積極參與討論，給文章提出一些寶貴的意見或者建議📝，合理的內容，我會采納更新博文，重新分享給大家，

🙏四連 關注🔎點贊👍收藏??留言📝

感謝大家的支持，用心寫博文分享給大家，你的支持（🔎點贊👍收藏??留言📝）是對我創作的最大幫助，
🍊微信公眾號：南北踏塵
🍊主頁地址：java_wxid
🍊社區地址：幕后大佬

給讀者大大的話

我本身是一個很普通的程式員，放在人堆里，除了與生俱來的🌹盛世美顏🌹、所剩不多的發量，就剩下180的大高個了，就是我這樣的一個人，默默堅持寫博文也有好多年了，有句老話說的好，🌕牛逼之前都是傻逼式的堅持🌕，希望自己可以通過大量的作品，時間的積累，個人魅力、運氣和時機，可以打造屬于自己的🌟技術影響力🌟，同時也希望自己可以成為一個🎄懂技術🎄，🎄懂業務🎄，🎄懂管理🎄的綜合型人才，作為專案架構路線的總設計師，掌控全域的🌕團隊大腦🌕，技術團隊中的🍊絕對核心🍊是我未來幾年不斷前進的目標，

面試資料

福利大放送，🎉歡迎關注🔎點贊👍收藏??留言📝，拜托了🙏，這對我真的很重要，
點擊：面試資料
提取碼：2021

200套PPT模板

福利大放送，🎉歡迎關注🔎點贊👍收藏??留言📝，拜托了🙏，這對我真的很重要，
點擊：200套PPT模板
提取碼：2021

提問的智慧

福利大放送，🎉歡迎關注🔎點贊👍收藏??留言📝，拜托了🙏，這對我真的很重要，
點擊：提問的智慧
提取碼：2021

Java開發學習路線

P5學習路線圖
P6學習路線圖
P7學習路線圖
P8學習路線圖

以上四張圖詳細介紹了作為Java開發作業者所需要具備的知識技能，同學們學廢了嘛，有想法系統學習的同學可以私聊我，🎉歡迎關注🔎點贊👍收藏??留言📝，
🍊博主：java_wxid
🍊博主：Java廖志偉
🍊社區：幕后大佬