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");         }     } }