import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer; // 引入ByteArrayDeserializer

import java.util.Properties;
import java.util.Arrays;

public class KafkaImageConsumerConfig {

    public static KafkaConsumer<String, byte[]> createConsumer(String bootstrapServers, String topic, String groupId) {
        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 關(guān)鍵修正：使用 ByteArrayDeserializer 處理 byte[] 類型的值
        prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 根據(jù)實(shí)際需求設(shè)置 MAX_POLL_RECORDS_CONFIG，默認(rèn)為 500
        // 如果設(shè)置為 1，每次 poll 只返回一條記錄，可能影響吞吐量
        // prop.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); // 暫時(shí)注釋或移除，詳見(jiàn)下一節(jié)

        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(prop);
        consumer.subscribe(Arrays.asList(topic));
        System.out.println("Kafka Consumer created and subscribed to topic: " + topic);
        return consumer;
    }

    public static void main(String[] args) {
        // 示例用法
        // KafkaConsumer<String, byte[]> consumer = createConsumer("localhost:9092", "image_topic", "image_group");
        // ... 后續(xù)消費(fèi)邏輯
    }
}

prop.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
// ...
ConsumerRecords<String,byte[]> records = dispatcher.consumer.poll(Duration.ofMillis(10));
int i = 0;
for (ConsumerRecord record : records) {
    // ...
    message_send[i]= java.util.Arrays.copyOf((byte[])record.value(), ((byte[])record.value()).length);

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

public class ImageConsumerProcessor {

    private final KafkaConsumer<String, byte[]> consumer;
    private final String topic;
    // 假設(shè)我們知道要接收的圖像總數(shù)，或者使用一個(gè)動(dòng)態(tài)列表
    private final int expectedNumberOfImages;
    private byte[][] receivedImages;
    private int imageCounter = 0; // 用于跟蹤已接收?qǐng)D像的數(shù)量和數(shù)組索引

    public ImageConsumerProcessor(KafkaConsumer<String, byte[]> consumer, String topic, int expectedImages) {
        this.consumer = consumer;
        this.topic = topic;
        this.expectedNumberOfImages = expectedImages;
        this.receivedImages = new byte[expectedImages][]; // 初始化數(shù)組
    }

    public void startConsuming() {
        System.out.println("Starting Image Consumption from topic: " + topic);

        try {
            // 持續(xù)消費(fèi)直到達(dá)到預(yù)期數(shù)量，或者根據(jù)業(yè)務(wù)邏輯退出
            while (imageCounter < expectedNumberOfImages) {
                // poll 方法會(huì)返回一個(gè) ConsumerRecords 集合，包含一個(gè)或多個(gè)記錄
                ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100)); // 設(shè)置合適的超時(shí)時(shí)間

                if (records.isEmpty()) {
                    System.out.println("No records found, polling again...");
                    // 可以添加短暫的休眠，避免空輪詢過(guò)于頻繁
                    // Thread.sleep(500);
                    continue;
                }

                System.out.println("Polling returned " + records.count() + " records.");

                for (ConsumerRecord<String, byte[]> record : records) {
                    if (imageCounter < expectedNumberOfImages) {
                        // 直接獲取 byte[] 類型的值
                        byte[] imageData = record.value();
                        receivedImages[imageCounter] = imageData; // 存儲(chǔ)圖像數(shù)據(jù)
                        System.out.println("Received image " + (imageCounter + 1) + ", offset: " + record.offset());
                        imageCounter++;
                    } else {
                        // 已經(jīng)接收到所有預(yù)期的圖像，可以考慮停止消費(fèi)或處理剩余消息
                        System.out.println("All expected images received. Skipping further processing.");
                        break; // 跳出當(dāng)前 records 循環(huán)
                    }
                }
                // 如果已達(dá)到預(yù)期數(shù)量，跳出外層 while 循環(huán)
                if (imageCounter >= expectedNumberOfImages) {
                    break;
                }
            }
        } catch (Exception e) {
            System.err.println("Error during consumption: " + e.getMessage());
            e.printStackTrace();
        } finally {
            consumer.close(); // 確保消費(fèi)者資源被關(guān)閉
            System.out.println("Consumer closed.");
        }
        System.out.println("Finished consuming images. Total received: " + imageCounter);
    }

    public byte[][] getReceivedImages() {
        return receivedImages;
    }

    public static void main(String[] args) {
        // 示例使用
        String bootstrapServers = "localhost:9092"; // 替換為你的Kafka服務(wù)器地址
        String topic = "image_topic"; // 替換為你的主題
        String groupId = "image_consumer_group"; // 替換為你的消費(fèi)者組ID
        int totalExpectedImages = 5; // 假設(shè)預(yù)期接收5張圖片

        KafkaConsumer<String, byte[]> consumer = KafkaImageConsumerConfig.createConsumer(bootstrapServers, topic, groupId);
        ImageConsumerProcessor processor = new ImageConsumerProcessor(consumer, topic, totalExpectedImages);
        processor.startConsuming();

        // 打印接收到的第一張圖像的大小作為驗(yàn)證
        if (processor.getReceivedImages() != null && processor.getReceivedImages().length > 0 && processor.getReceivedImages()[0] != null) {
            System.out.println("Size of first received image: " + processor.getReceivedImages()[0].length + " bytes");
        }
    }
}