Java Kafka消费者接收图像数据：从配置到处理的完整指南

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

import java.util.Properties;

public class KafkaConsumerConfig {

    public static Properties createConsumerProperties(String bootstrapServers, String groupId, String topic) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 关键：将值反序列化器设置为 ByteArrayDeserializer
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        // 首次启动或无有效偏移量时，从最早的可用偏移量开始消费
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 控制每次 poll 返回的最大记录数，根据实际需求调整
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10); 
        return props;
    }
}

import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class ImageConsumer {

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

    public ImageConsumer(String bootstrapServers, String groupId, String topic) {
        Properties props = new Properties();
        props.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        props.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        props.setProperty(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // 每次 poll 最多获取的记录数，根据实际吞吐量和内存情况调整
        props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 5); 

        this.consumer = new KafkaConsumer<>(props);
        this.topic = topic;
        // 订阅主题，此操作应在消费循环开始前执行一次
        consumer.subscribe(Arrays.asList(topic));
    }

    public void startConsuming() {
        System.out.println("Starting Kafka Image Consumer for topic: " + topic);
        try {
            while (true) { // 持续消费
                // poll 方法定期拉取消息，参数为超时时间
                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.printf("Offset = %d, Key = %s, Value Size = %d bytes%n", 
                                      record.offset(), record.key(), record.value().length);

                    // 获取图像的字节数组
                    byte[] imageData = record.value();

                    // 在这里处理图像数据，例如保存到文件或转换为BufferedImage
                    processImageBytes(imageData, record.offset());
                }
            }
        } catch (Exception e) {
            System.err.println("Error during consumption: " + e.getMessage());
            e.printStackTrace();
        } finally {
            // 确保消费者在退出时关闭，释放资源
            consumer.close();
            System.out.println("Kafka Consumer closed.");
        }
    }

    private void processImageBytes(byte[] imageData, long offset) {
        // 示例：简单地打印字节数组长度，实际应用中会进行图像解析
        System.out.println("Processing image data from offset " + offset + ", length: " + imageData.length);
        // 实际应用中，可以使用javax.imageio.ImageIO或OpenCV等库将byte[]转换为图像对象
        // try {
        //     ByteArrayInputStream bis = new ByteArrayInputStream(imageData);
        //     BufferedImage image = ImageIO.read(bis);
        //     if (image != null) {
        //         System.out.println("Image successfully read: " + image.getWidth() + "x" + image.getHeight());
        //         // ImageIO.write(image, "png", new File("received_image_" + offset + ".png"));
        //     } else {
        //         System.err.println("Could not read image from bytes.");
        //     }
        // } catch (IOException e) {
        //     System.err.println("Error processing image: " + e.getMessage());
        // }
    }

    public static void main(String[] args) {
        String bootstrapServers = "localhost:9092"; // 替换为你的Kafka服务器地址
        String groupId = "image_consumer_group";
        String topic = "image_topic"; // 替换为你的Kafka主题

        ImageConsumer consumerApp = new ImageConsumer(bootstrapServers, groupId, topic);
        consumerApp.startConsuming();
    }
}