springboot实现读kafka数据写入文件并按指定行数分割文件

青旅半醒 2021-09-23 23:46 453阅读 0赞

## 1、需求描述 ##

大致需求是这样的，kafka作为数据缓存通道，接收前端设备上传的数据，现在需要实现读取kafka的数据，再保存到文件，但是不能全部写入一个文件，会导致一个文件过大，需要按照指定行数分割文件，按行数是为了方便核对数据量，前端上传了多少数据，kafka接收了多少数据，最终写入文件多少数据。

另外需要说明传入的数据是一个复杂的json字符串，嵌套了很多内容，而且还有图片的base64字符串。

## 2、实现方法 ##

方式一：

采用spark streaming读kafka数据写本地文件，每个时间间隔一个小文件，会导致大量的小文件存在，这样是有问题的，所以没有深入研究

方式二：

采用springboot java代码读kafka写本地文件，可以实现需求，而且更加灵活，可以灵活解析复杂的json，也可以把base64图片保存为图片文件。

方式三：

采用kettle读kafka，可以指定文件命名规则，带日期的，指定每个文件多少行数据，方便数据量核对。达到指定行数才会生成新文件，所以可能指定的文件行数过大，内存不够的话可能会存在问题！解析复杂的json存在问题，用组件拼接太繁琐，尝试了一下放弃了，另外base64图片转为文件也是个问题。。

方式四：

采用sdc读kafka写文件，可以指定文件命名规则，带日期，指定每个文件多少行数据，或者每个文件大小，建议使用记录数分割，达到指定行数才会生成新文件，未达到指定数量会写入临时文件，达到数量之后会转换成真实文件。任务停止会把临时文件保存为正式文件。sdc指的是streamsets datacollector，这个也和kettle类似，组件化的，解析复杂json存在困难。

综合考虑，最终使用方式二来实现功能。

## 3、核心代码 ##

为了方便采用springboot框架，使用@KafkaListener(topics = \{"test"\})监听指定的topic数据，然后接收到数据后，进行解析并存入文件，通过几个计数参数控制文件的指定条数。具体实现如下：

@Component
    public class ReceiveSaveSplit {
        private final org.slf4j.Logger log = LoggerFactory.getLogger(getClass());
        Long exportNum=10L;
    
        Long countNum=0L;
    
        Long lineNum=0L;
    
        int i = 0;
    
        String pathname="/Users/kafka-part"+i+".txt";
        OutputStreamWriter osw;
    
        {
            try {
                osw = getOutputFile(pathname);
            } catch (IOException e) {
                e.printStackTrace();
    
            }
        }
    
        @KafkaListener(topics = {"test"})
        public void listen(ConsumerRecord<?, ?> record) throws IOException{
            log.info("offset is : " + record.offset());
            log.info("开始消费消息"+new Date());
    
                Optional<?> kafkaMessage = Optional.ofNullable(record.value());
                if (kafkaMessage.isPresent()) {
                    Object message = kafkaMessage.get();
                    if (message != null)
                    {
                        lineNum++;
                        countNum++;
    
                        System.out.println("offset："+record.offset()+"，countNum："+countNum+"，lineNum："+lineNum+"，message："+message.toString());
                        //StaticData.messageBlockingQueue.add(message.toString());
                        if(lineNum%exportNum==0){
                            osw.close();
                            i++;
                            osw= getOutputFile("/Users/kafka-part"+i+".txt");
                            lineNum=0L;
                        }
                        savePeople(message.toString(),osw);
                    }
                }
        }
    /**
         *  追加方式写数据
         * @param pathname
         * @return
         * @throws IOException
         */
        private static OutputStreamWriter getOutputFile(String pathname) throws IOException {
            File file=new File(pathname);
            FileOutputStream fos = null;
            if(!file.exists()){
                file.createNewFile();//如果文件不存在，就创建该文件
                fos = new FileOutputStream(file);//首次写入获取
            }else{
                //如果文件已存在，那么就在文件末尾追加写入
                fos = new FileOutputStream(file,true);//这里构造方法多了一个参数true,表示在文件末尾追加写入
            }
            OutputStreamWriter osw = new OutputStreamWriter(fos, "UTF-8");//指定以UTF-8格式写入文件
            return osw;
        }

## 4、反思 ##

一开始犯了一个错误，把计数变量和文件创建放到listen方法里面了，导致数据计数每次都是0，后来把这部分代码放到外面就可以了，自己对代码的逻辑掌握还不够熟练，需要多学习改进。