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;
import java.time.Duration;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;

public class ImageConsumer {

    private KafkaConsumer<String, byte[]> consumer;
    private String topic;

    public ImageConsumer(String bootstrapServers, String consumerId, String topic) {
        Properties prop = new Properties();
        // Kafka集群的连接地址
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // 键的反序列化器，通常使用StringDeserializer
        prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 值的反序列化器，对于图像等二进制数据，必须使用ByteArrayDeserializer
        prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        // 消费者组ID，用于管理消费者群组和偏移量
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, consumerId);
        // 当没有初始偏移量或当前偏移量无效时，如何重置偏移量
        // "earliest"表示从最早的可用偏移量开始消费
        prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 每次poll调用返回的最大记录数
        // 默认值为500，如果设置为1，每次poll只会返回一条记录
        prop.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 500); // 建议设置为合理的值，或不设置使用默认值

        this.consumer = new KafkaConsumer<>(prop);
        this.topic = topic;
        // 订阅指定主题，此操作通常在消费者初始化后执行一次
        consumer.subscribe(Arrays.asList(topic));
    }

    // 示例：消费循环
    public void startConsuming() {
        System.out.println("Starting Kafka Consumer for topic: " + topic);
        try {
            while (true) { // 持续消费
                // 轮询Kafka获取消息，设置超时时间
                ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100));
                if (records.isEmpty()) {
                    // System.out.println("No records received, polling again...");
                    continue;
                }

                System.out.println("Received " + records.count() + " records.");
                for (ConsumerRecord<String, byte[]> record : records) {
                    System.out.println("Offset: " + record.offset() + ", Key: " + record.key() + ", Value length: " + record.value().length);
                    // 处理接收到的图像数据 (byte[])
                    byte[] imageData = record.value();
                    // 这里可以添加图像处理逻辑，例如保存到文件或进行进一步分析
                    // 例如：ImageIO.write(ImageIO.read(new ByteArrayInputStream(imageData)), "jpg", new File("image_" + record.offset() + ".jpg"));
                }
            }
        } catch (Exception e) {
            System.err.println("Error during consumption: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 关闭消费者，释放资源
            consumer.close();
            System.out.println("Kafka Consumer closed.");
        }
    }

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092"; // 替换为你的Kafka服务器地址
        String consumerId = "image-consumer-group";
        String topicName = "image-topic"; // 替换为你的Kafka主题名

        ImageConsumer imageConsumer = new ImageConsumer(bootstrapServers, consumerId, topicName);
        imageConsumer.startConsuming();
    }
}

prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());

prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());