Java實作kafka集群的生產者與消費者案例（小白)-有解無憂

在初學kafkaAPI的時候，查閱大量大佬文章，發現開始都是要進行構建kafka依賴，但是由于博主對于maven依賴不熟悉以及對idea的操作略少，故在進行構建maven依賴，寫入pom.xml的時候出現錯誤，讓博主很是煩心，但是本文的重點不在于糾結如何寫依賴，關于maven依賴問題，博主會在有空的時候去研究，這個不急（是不是感覺博主好菜啊~~~~~）

因此本文直接把Linux中的kafka集群的j有關jar包，直接copy了出來，這樣也能實作呼叫kafkaAPI，實作生產者與消費者案例（嘿嘿，博主聰明吧），借此告訴大家一個道理：解決問題要尋求多渠道解決，不要死磕在一條道上，好了，話不多說，直接進入主題！！！

首先，找到任一臺kafka節點的根目錄下，如博主的kafka路徑為如下：

ls查看，我們發現libs目錄下有很多jar包：

本文所用到的jar包就在其中，故下一步我們就要把libs檔案夾添加到我們Window上的idea依賴上，至于如何把libs檔案夾傳到Windows上，想必大家都有辦法的吧，利用VM虛擬機的共享檔案夾或者xftp等都可以實作Linux檔案傳到Window,博主用的是MobaXterm，故可以直接把libs檔案夾拖到Windows，但是利用其它軟體的話，需要提前壓縮libs檔案夾，然后再傳到Windows!!!

接著還要修改kafka集群的埠，每個節點(博主的集群為：master,slave1,slave2)都要設定，進入kafka的config目錄下，vim server.properties，先設定master節點：

在如下地方添加：port=9092

同理，另外兩臺節點（slave1，slave2）也同樣設定：

slave1：

slave2：

然后在每個節點上啟動kafka集群（既然要設定了埠號，就必須先啟用kafka集群，方可讓Java實作kafka生產者跟消費者功能，

---------------------------------------------------------------------------------------------------------------------------------------------------------------------

接下來進行具體代碼實作環節：

首先找到提前傳好的libs檔案（如果你的libs檔案夾沒有解壓的話，必須先解壓），

在idea中創建kafka工程，依次點擊工具列的File-->New-->Project，

然后創建maven

填寫點擊Next,設定工程名字，博主的是Kafka，然后點擊Finish

接著把剛才傳過來libs檔案夾放到這個工程，工程路徑不清楚的話，參考上面的路徑，最后情況如下：

到這里就可以寫代碼了，（再次確保虛擬機中的kafka集群啟動成功），在src下創建生產者類MyProducer跟消費者類

生產者代碼如下：

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

/**
 *
 * Title: MyProducer
 * Description:
 *  kafka生產者 demo
 * Version:2.2.2
 * @author haigui
 * @date 2020/10/2
 */
public class MyProducer implements Runnable {

    private final KafkaProducer<String, String> producer;
    private final String topic;
    public MyProducer(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        props.put("acks", "all");
        props.put("retries", 0);
        props.put("batch.size", 16384);
        props.put("key.serializer", StringSerializer.class.getName());
        props.put("value.serializer", StringSerializer.class.getName());
        this.producer = new KafkaProducer<String, String>(props);
        this.topic = topicName;
    }

    public void run() {
        int messageNo = 1;
        try {
            for(;;) {
                String messageStr="你好，這是第"+messageNo+"條資料";
                producer.send(new ProducerRecord<String, String>(topic, "Message", messageStr));
                //生產了100條就列印
                if(messageNo%100==0){
                    System.out.println("發送的資訊:" + messageStr);
                }
                //生產1000條就退出
                if(messageNo%1000==0){
                    System.out.println("成功發送了"+messageNo+"條");
                    break;
                }
                messageNo++;
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            producer.close();
        }
    }

    public static void main(String args[]) {
        MyProducer test= new MyProducer("KAFKA_TEST");
        Thread thread = new Thread(test);
        thread.start();
    }
}

消費者代碼：

import java.util.Arrays;
import java.util.Properties;
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;


/**
 *
 * Title: MyConsumer
 * Description:
 *  kafka消費者 demo
 * Version:2.2.2
 * @author haigui
 * @date 2020/10/2
 */
public class    MyConsumer implements Runnable {
    private final KafkaConsumer<String, String> consumer;
    private ConsumerRecords<String, String> msgList;
    private final String topic;
    private static final String GROUPID = "groupA";

    public MyConsumer(String topicName) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "master:9092,slave1:9092,slave2:9092");
        props.put("group.id", GROUPID);
        props.put("enable.auto.commit", "true");
        props.put("auto.commit.interval.ms", "1000");
        props.put("session.timeout.ms", "30000");
        props.put("auto.offset.reset", "earliest");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
        this.consumer = new KafkaConsumer<String, String>(props);
        this.topic = topicName;
        this.consumer.subscribe(Arrays.asList(topic));
    }

    public void run() {
        int messageNo = 1;
        System.out.println("---------開始消費---------");
        try {
            for (;;) {
                msgList = consumer.poll(1000);
                if(null!=msgList&&msgList.count()>0){
                    for (ConsumerRecord<String, String> record : msgList) {
                        //消費100條就列印 ,但列印的資料不一定是這個規律的
                        if(messageNo%100==0){
                            System.out.println(messageNo+"=======receive: key = " + record.key() + ", value = " + record.value()+" offset==="+record.offset());
                        }
                        //當消費了1000條就退出
                        if(messageNo%1000==0){
                            break;
                        }
                        messageNo++;
                    }
                }else{
                    Thread.sleep(1000);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            consumer.close();
        }
    }
    public static void main(String args[]) {
            MyConsumer test1 = new MyConsumer("KAFKA_TEST");
        Thread thread1 = new Thread(test1);
        thread1.start();
    }
}

其中在生產者demo跟消費者demo中的配置“bootstrap.servers”,要跟你kafka節點的ip或者hostname對應，如下圖

有關配置說明，博主不在這解釋，有興趣的同學可以去查閱資料（本文主要讓小白掌握如何讓idea跑kafka集群的生產者跟消費者）

最后便是run啦，先運行消費者demo，如下：

然后再是運行生產者demo，如下：

最后回過頭看消費者終端，發現接收了資訊，如下：

好啦，最后消費者出現上面的資訊，代表Java實作了本文目標：生產者與消費者java代碼實作實體，

感謝大家的閱讀，有啥問題或者建議下方評論指出哦，有幫助請點點贊哈！！！

轉載請註明出處，本文鏈接：https://www.uj5u.com/qianduan/155197.html

標籤：其他

上一篇：SpringCloud使用Eureka作為服務治理中心

下一篇：ES6封裝MongoDB的CRUD