Java Kafka消费者接收图像数据：反序列化与高效处理指南

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());
        // 关键修正：使用 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");
        // 根据实际需求设置 MAX_POLL_RECORDS_CONFIG，默认为 500
        // 如果设置为 1，每次 poll 只返回一条记录，可能影响吞吐量
        // prop.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); // 暂时注释或移除，详见下一节

        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");
        // ... 后续消费逻辑
    }
}

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;
    // 假设我们知道要接收的图像总数，或者使用一个动态列表
    private final int expectedNumberOfImages;
    private byte[][] receivedImages;
    private int imageCounter = 0; // 用于跟踪已接收图像的数量和数组索引

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

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

        try {
            // 持续消费直到达到预期数量，或者根据业务逻辑退出
            while (imageCounter < expectedNumberOfImages) {
                // poll 方法会返回一个 ConsumerRecords 集合，包含一个或多个记录
                ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100)); // 设置合适的超时时间

                if (records.isEmpty()) {
                    System.out.println("No records found, polling again...");
                    // 可以添加短暂的休眠，避免空轮询过于频繁
                    // 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; // 存储图像数据
                        System.out.println("Received image " + (imageCounter + 1) + ", offset: " + record.offset());
                        imageCounter++;
                    } else {
                        // 已经接收到所有预期的图像，可以考虑停止消费或处理剩余消息
                        System.out.println("All expected images received. Skipping further processing.");
                        break; // 跳出当前 records 循环
                    }
                }
                // 如果已达到预期数量，跳出外层 while 循环
                if (imageCounter >= expectedNumberOfImages) {
                    break;
                }
            }
        } catch (Exception e) {
            System.err.println("Error during consumption: " + e.getMessage());
            e.printStackTrace();
        } finally {
            consumer.close(); // 确保消费者资源被关闭
            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服务器地址
        String topic = "image_topic"; // 替换为你的主题
        String groupId = "image_consumer_group"; // 替换为你的消费者组ID
        int totalExpectedImages = 5; // 假设预期接收5张图片

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

        // 打印接收到的第一张图像的大小作为验证
        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");
        }
    }
}