Spring Kafka大视频逐字节分片传输与消费端重组实现方案

云上老周

2026-8-27

现有实现代码

Kafka生产者配置

import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaProducerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootStrapServers;

    public Map<String, Object> producerConfig() {
        HashMap<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.MAX_REQUEST_SIZE_CONFIG, "20971520");
        return props;

    }

    @Bean
    public ProducerFactory<String, String> producerFactory() {
        return new DefaultKafkaProducerFactory<String, String>(producerConfig());
    }

    @Bean
    public KafkaTemplate<String, String> kafkaTemplate(ProducerFactory<String, String> producerFactory) {
        return new KafkaTemplate<String, String>(producerFactory);
    }
}

Kafka消费者配置

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConsumerConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootStrapServers;

    public Map<String, Object> consumerConfig() {
        HashMap<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootStrapServers);
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        // 修正原有配置和生产者序列化方式不匹配的问题，统一使用String反序列化
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        return props;
    }

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<String, String>(consumerConfig());
    }

    @Bean
    public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> factory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

Kafka消息发送服务（原有实现）

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

@Service
public class KafkaSender {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    String kafkaTopic = "testTopic";
    public void send() {
        byte[] array = null;

        try {
            array = Files.readAllBytes(Paths.get("Test.webm"));
            String kafkaTopic = "testTopic";
            String encoded = java.util.Base64.getEncoder().encodeToString(array);
            kafkaTemplate.send(kafkaTopic, encoded);

        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}

Kafka消息监听器（原有实现）

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Base64;

@Component
public class Listener {

    @KafkaListener(topics = "testTopic", groupId = "foo")
    public void listenGroupFoo(String message){
        byte[] decoded = java.util.Base64.getDecoder().decode(message);
        try {
            FileOutputStream out;
            out = new FileOutputStream("video1.mp4");
            out.write(decoded);
            out.close();
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }
}

大文件分片传输可行方案

注意：不要按逐字节分片，1GB文件对应10亿条消息，网络开销、Kafka集群压力都会大到完全不可用，常规做法是按1MB~15MB的固定大小切分，兼顾可靠性和性能。
核心思路是给每个文件分片带上唯一标识、序号、总分片数三个元数据，生产者按固定大小切分流式发送，消费者收到分片后按序号直接写入临时文件对应位置，全程避免把整个文件加载到内存导致OOM。

核心改造点

统一消息结构
用JSON作为消息序列化格式，包含4个核心字段：
- fileId：文件唯一标识，用UUID生成，同一个文件的所有分片共用同一个fileId，发送时用fileId作为Kafka消息key，保证同文件分片进入同一个分区、按发送顺序消费
- chunkIndex：当前分片序号，从0开始计数
- totalChunks：当前文件总分片数
- content：当前分片内容的Base64编码字符串
改造生产者发送逻辑
抛弃一次性读取全量文件到内存的做法，用文件流边读边切边发：
1. 定义分片大小，考虑Base64编码后体积会膨胀1/3，你单条消息上限配置为20MB，原始分片大小设为10MB即可，留足冗余
2. 计算文件总长度，算出总分片数
3. 生成唯一fileId
4. 循环读取文件，每次读满分片大小的字节，编码后封装成消息发送
  改造后的发送代码示例：

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import java.io.FileInputStream;
import java.io.IOException;
import java.util.Base64;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

@Service
public class KafkaSender {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    // 分片大小设为10MB
    private static final int CHUNK_SIZE = 10 * 1024 * 1024;
    private final String kafkaTopic = "testTopic";
    private final ObjectMapper objectMapper = new ObjectMapper();

    public void sendLargeFile(String filePath) throws IOException {
        FileInputStream fis = new FileInputStream(filePath);
        // 生成本次传输的文件唯一ID
        String fileId = UUID.randomUUID().toString();
        // 计算总分片数
        long fileSize = fis.getChannel().size();
        int totalChunks = (int) Math.ceil((double) fileSize / CHUNK_SIZE);
        byte[] buffer = new byte[CHUNK_SIZE];
        int chunkIndex = 0;
        int readLen;
        while ((readLen = fis.read(buffer)) != -1) {
            // 处理最后一片不足CHUNK_SIZE的情况
            byte[] chunkBytes;
            if (readLen < CHUNK_SIZE) {
                chunkBytes = new byte[readLen];
                System.arraycopy(buffer, 0, chunkBytes, 0, readLen);
            } else {
                chunkBytes = buffer;
            }
            // 封装消息
            Map<String, Object> msg = new HashMap<>();
            msg.put("fileId", fileId);
            msg.put("chunkIndex", chunkIndex);
            msg.put("totalChunks", totalChunks);
            msg.put("content", Base64.getEncoder().encodeToString(chunkBytes));
            // 用fileId作为消息key，保证同文件分片进入同一分区有序
            kafkaTemplate.send(kafkaTopic, fileId, objectMapper.writeValueAsString(msg));
            chunkIndex++;
        }
        fis.close();
    }
}

改造消费者重组逻辑
抛弃收到消息就直接写文件的逻辑，用临时文件随机写的方式重组，避免大文件占满内存：
1. 维护一个线程安全的缓存，记录每个fileId对应的临时文件通道、已收到的分片数
2. 收到分片消息后解析元数据，把分片内容解码后写到临时文件的对应偏移位置
3. 每收到一个分片就计数，当计数等于总分片数时，关闭文件通道，把临时文件重命名为目标文件，清理缓存
  改造后的监听器代码示例：

import com.fasterxml.jackson.databind.ObjectMapper;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.util.Base64;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class Listener {
    private static final int CHUNK_SIZE = 10 * 1024 * 1024;
    private final ObjectMapper objectMapper = new ObjectMapper();
    // 缓存每个fileId对应的临时文件操作对象和已接收分片数
    private final ConcurrentHashMap<String, FileTransferContext> transferCache = new ConcurrentHashMap<>();

    @KafkaListener(topics = "testTopic", groupId = "foo")
    public void listenChunk(String message) throws IOException {
        Map msg = objectMapper.readValue(message, Map.class);
        String fileId = (String) msg.get("fileId");
        int chunkIndex = (int) msg.get("chunkIndex");
        int totalChunks = (int) msg.get("totalChunks");
        byte[] content = Base64.getDecoder().decode((String) msg.get("content"));

        // 初始化当前文件的传输上下文
        transferCache.computeIfAbsent(fileId, k -> {
            try {
                // 创建临时文件
                File tempFile = File.createTempFile("kafka_transfer_", ".tmp");
                RandomAccessFile raf = new RandomAccessFile(tempFile, "rw");
                return new FileTransferContext(raf, tempFile, 0);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        });

        FileTransferContext context = transferCache.get(fileId);
        // 加锁保证同文件写入顺序正确
        synchronized (context) {
            context.raf.seek((long) chunkIndex * CHUNK_SIZE);
            context.raf.write(content);
            context.receivedChunkCount++;
            // 判断是否收齐所有分片
            if (context.receivedChunkCount == totalChunks) {
                context.raf.close();
                // 重命名临时文件为最终视频文件
                File targetFile = new File("video1.mp4");
                context.tempFile.renameTo(targetFile);
                // 清缓存
                transferCache.remove(fileId);
            }
        }
    }

    // 传输上下文内部类
    private static class FileTransferContext {
        RandomAccessFile raf;
        File tempFile;
        int receivedChunkCount;

        public FileTransferContext(RandomAccessFile raf, File tempFile, int receivedChunkCount) {
            this.raf = raf;
            this.tempFile = tempFile;
            this.receivedChunkCount = receivedChunkCount;
        }
    }
}

注意事项

可以根据自己的网络情况、Kafka集群配置调整分片大小，只要单条消息大小不超过集群配置的message.max.bytes和replica.fetch.max.bytes上限即可
生产环境可以加超时清理逻辑，比如某个fileId的分片超过1小时没传完，就删除临时文件、清理缓存，避免垃圾文件占磁盘空间
如果需要更高的传输性能，可以去掉Base64编码，直接用字节数组序列化方式传原始分片字节，能减少1/3的网络传输量和编解码开销

内容的提问来源于stack exchange，提问作者Prakash Singh

Spring Kafka大视频逐字节分片传输与消费端重组实现方案

Kafka生产者配置 #

Kafka消费者配置 #

Kafka消息发送服务（原有实现） #

Kafka消息监听器（原有实现） #

核心改造点 #