如果沒有安裝kafka，可以參考我的另一篇文章

先來個生產者的

import java.util.Properties;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;



public class kafkaProducer {
    public static String topic = "test";
    public static Producer<String, String> producer;

    public static void main(String[] args) throws InterruptedException {
        Properties props = new Properties();
        props.put("bootstrap.servers", "127.0.0.1:9092");
        // 0不需要等待任何確認收到的資訊(效能最佳) 1至少要等待leader已經成功將資料寫入本地log(中等) all所有備份都成功寫入日誌(最強保住不丟)
        props.put("acks", "1");
        // 訊息傳送最大嘗試次數
        props.put("retries", 0);
        // 批量處理的訊息的大小 小於此size會進行批量打包傳送 設定過大會佔用太多記憶體 太小降低吞吐
        props.put("batch.size", 16384);
        // 批量傳送訊息的延遲為5ms
        props.put("linger.ms", 5);
        // 快取資料的記憶體大小 如果產生資料速度大於傳送速度 並且快取超過這個大小 則丟擲異常
        props.put("buffer.memory", 33554432);
        // key和value的編碼序列化方法
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        producer = new KafkaProducer<String, String>(props);
        for (int i = 0; i < 10000; i++) {
            Thread.sleep(1);
            ProducerRecord<String, String> producerRecord =
                    new ProducerRecord<String, String>(topic, "", "topic向你發來慰問" + i);
            producer.send(producerRecord);
        }
    }
}
 

下面是消費者的

package com.kafka.mykafka;

import java.util.Arrays;
import java.util.Properties;
import java.util.concurrent.ArrayBlockingQueue;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class kafkaConsumer extends Thread {

    public static String topic = "test";
    /** 訊息儲存佇列 */
    public static ArrayBlockingQueue<ConsumerRecord<String, String>> recordQueue =
            new ArrayBlockingQueue<ConsumerRecord<String, String>>(10000);
    public static KafkaConsumer<String, String> consumer;

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("bootstrap.servers", "127.0.0.1:9092");
        // 分組。一條訊息只會被同組的一個consume消費
        props.put("group.id", "test");
        // 自動提交offset; false的話還需要手動呼叫consumer.commitSync()
        props.put("enable.auto.commit", "true");
        // 自動提交offset的時間間隔 太短效能會差，太長如果crash重啟後可能會重複消費的多(前提是訊息在kafka伺服器上還在)
        props.put("auto.commit.interval.ms", "1000");
        // session 過期時間 一旦心跳連線不上，超過這個時間就會連結斷開
        props.put("session.timeout.ms", "30000");
        // # 重複消費,設定為latest,如遊戲中的聊天等
        props.put("auto.offset.reset", "latest");
        // key和value的解碼序列化方法
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(props);
        // 繫結topic
        consumer.subscribe(Arrays.asList(topic));
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(1000);
            System.out.println("收到一次kafka訊息,size:" + records);
            for (ConsumerRecord<String, String> record : records) {
                recordQueue.offer(record);
                System.out.printf("offset = %d, key = %s, value = %s\n", record.offset(),
                        record.key(), record.value());
            }
            System.out.println("recordQueue" + recordQueue.size());
        }
    }
}
 

然後再執行即可搞定