Kafka Java消费者接收图像数据：类型转换与多记录处理实践

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;

public class ImageConsumerConfig {

    public KafkaConsumer<String, byte[]> createConsumer(String bootstrapServers, String topic, String consumerId) {
        Properties prop = new Properties();
        prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
        // 关键修正：将值反序列化器设置为ByteArrayDeserializer
        prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class.getName());
        prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, consumerId);
        prop.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        // prop.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1); // 暂时注释或根据需求调整，下文会详细讨论

        // 消费者声明的泛型类型也必须与反序列化器匹配
        KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(prop);
        // consumer.subscribe(Arrays.asList(topic)); // 订阅可以在创建后进行

        return consumer;
    }
}

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

public class ImageConsumerLogic {

    // 假设 dispatcher.consumer 已正确初始化
    // 假设 dispatcher.AcceptedNumberJobs 和 dispatcher.queue_size 是用于控制循环的计数器
    // 为了示例清晰，这里简化了 dispatcher 的使用

    public void consumeImages(KafkaConsumer<String, byte[]> consumer, String topic, int expectedRecords) {
        List<byte[]> receivedImages = new ArrayList<>(); // 使用列表动态存储接收到的图像

        System.out.println("Starting Consuming");
        // 订阅主题，通常在消费者创建后订阅一次即可
        consumer.subscribe(Collections.singletonList(topic)); 

        // 示例循环条件：直到接收到足够数量的图像或达到某个退出条件
        while (receivedImages.size() < expectedRecords) { 
            System.out.println("Polling for records...");
            ConsumerRecords<String, byte[]> records = consumer.poll(Duration.ofMillis(100)); // 增加poll超时时间以等待更多消息

            if (records.isEmpty()) {
                System.out.println("No records received in this poll. Waiting...");
                continue; // 如果没有记录，继续下一次poll
            }

            System.out.println("Received " + records.count() + " records.");
            for (ConsumerRecord<String, byte[]> record : records) {
                // 直接处理或存储接收到的字节数组
                byte[] imageData = record.value();
                receivedImages.add(imageData); // 将图像数据添加到列表中

                // 打印一些信息以验证
                System.out.println("Received image with size: " + imageData.length + " bytes from offset: " + record.offset());
                // 根据实际需求，这里可以进一步处理 imageData，例如保存到文件、显示等
            }
            // 提交偏移量，确保下次从正确的位置开始消费
            consumer.commitSync(); 
        }
        System.out.println("Finished consuming. Total images received: " + receivedImages.size());
        // 此时 receivedImages 列表中包含了所有接收到的图像数据
    }
}