`

Java nio导入csv对账文件

阅读更多

公司业务上传支付宝、微信交易记录,并和系统进行对账 功能

个人使用了,java 的nio  和 多线程进行扫描文件 并装载bean对象,具体代码如下:

附件有对应的工具类,一直上传失败,放到百度云盘,有需要的可以取下载:链接: https://pan.baidu.com/s/1y-I36iUAQbx2_Ss1Ih8ASQ 提取码: meng 

 

@RequestMapping(value = "checkReconciliationData/{channel}", method = RequestMethod.POST)
public CResponse checkReconciliationDataNew(@RequestParam("file") MultipartFile file, @PathVariable String channel)throws IOException {
    if (file != null && !file.isEmpty()) {
        String fileName = file.getOriginalFilename();
        UploadPayDataRequest uploadPayDataRequest = new UploadPayDataRequest();
        uploadPayDataRequest.setFileName(fileName);
        List<UploadPayDataEntity> uploadPayDataEntities = uploadPayDataService.selectInfoByList(uploadPayDataRequest);
        if (uploadPayDataEntities.size() > 0) {
            return CResponse.error(ResultCodeEnum.ERROR_DATA_FORMAT, "文件已存在,无需重复上传!");
        }
        String keyName = MD5.MD5Encode(fileName);
        System.out.println("keyName:" + keyName);
        String keyValue = jedisCluster.get(keyName);
        if (keyValue == null) {
            jedisCluster.set(keyName, "PROCESSING");//文件处理中
            ExecutorService executorService = Executors.newSingleThreadExecutor();
            UploadPayDataDto uploadPayDataDto = new UploadPayDataDto();
            executorService.execute(() -> {
                try {
                    CommonsMultipartFile cf = (CommonsMultipartFile) file;
                    DiskFileItem fi = (DiskFileItem) cf.getFileItem();
                    File f = fi.getStoreLocation();
                    BigFileReader.Builder builder = new BigFileReader.Builder(f, line -> {
                        if (line.indexOf("渠道") == -1) {
                            String[] strs = line.split(",");
                            String json = JSONObject.toJSONString(setItem(strs, channel, fileName));
                            jedisCluster.lpush(fileName, json);
                        }
                    });
                    BigFileReader bigFileReader = builder
                            .threadPoolSize(10)
                            .charset(Charset.forName("GBK"))
                            .bufferSize(1024).build();
                    bigFileReader.start(jedisCluster, keyName);
                    upload(file);
                } catch (Exception e) {
                    log.error("异步操作文件内容失败,原因:" + e.getMessage());
                    jedisCluster.del(keyName);
                }
            });
            BufferedInputStream bis = new BufferedInputStream(file.getInputStream());
            BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(bis, "GBK"), 30 * 1024 * 1024);//10M缓存
            uploadPayDataDto.setTotalNum(bufferedReader.lines().count());
            uploadPayDataDto.setUrl(ALIYUN_IMAGE_URL + fileName);
            uploadPayDataDto.setFileName(fileName);
            return CResponse.success(uploadPayDataDto);
        } else {
            return CResponse.error(ResultCodeEnum.ERROR_DATA_FORMAT, "文件处理中");
        }
    } else {
        return CResponse.error(ResultCodeEnum.ERROR_DATA_FORMAT, "文件数据为空");
    }
}



public class BigFileReader {
    private int threadPoolSize;
    private Charset charset;
    private int bufferSize;
    private IFileHandle handle;
    private ExecutorService executorService;
    private long fileLength;
    private RandomAccessFile rAccessFile;
    private Set<StartEndPair> startEndPairs;
    private CyclicBarrier cyclicBarrier;
    private AtomicLong counter = new AtomicLong(0);


    public BigFileReader(File file, IFileHandle handle, Charset charset, int bufferSize, int threadPoolSize) {
        this.fileLength = file.length();
        this.handle = handle;
        this.charset = charset;
        this.bufferSize = bufferSize;
        this.threadPoolSize = threadPoolSize;
        try {
            this.rAccessFile = new RandomAccessFile(file, "r");
        } catch (FileNotFoundException e) {
            e.printStackTrace();
        }
        this.executorService = Executors.newFixedThreadPool(threadPoolSize);
        startEndPairs = new HashSet<>();
    }

    public void start(JedisCluster jedisCluster, String fileName) {
        long everySize = this.fileLength / this.threadPoolSize;
        try {
            calculateStartEnd(0, everySize);
        } catch (IOException e) {
            e.printStackTrace();
            return;
        }

        final long startTime = System.currentTimeMillis();
        cyclicBarrier = new CyclicBarrier(startEndPairs.size(), () -> {
            System.out.println("use time: " + (System.currentTimeMillis() - startTime));
            System.out.println("all line: " + counter.get());
            System.out.println(fileName);
            jedisCluster.set(fileName, "SUCCESS");
            shutdown();
        });
        for (StartEndPair pair : startEndPairs) {
            System.out.println("分配分片:" + pair);
            this.executorService.execute(new SliceReaderTask(pair));
        }
    }

    private void calculateStartEnd(long start, long size) throws IOException {
        if (start > fileLength - 1) {
            return;
        }
        StartEndPair pair = new StartEndPair();
        pair.start = start;
        long endPosition = start + size - 1;
        if (endPosition >= fileLength - 1) {
            pair.end = fileLength - 1;
            startEndPairs.add(pair);
            return;
        }

        rAccessFile.seek(endPosition);
        byte tmp = (byte) rAccessFile.read();
        while (tmp != '\n' && tmp != '\r') {
            endPosition++;
            if (endPosition >= fileLength - 1) {
                endPosition = fileLength - 1;
                break;
            }
            rAccessFile.seek(endPosition);
            tmp = (byte) rAccessFile.read();
        }
        pair.end = endPosition;
        startEndPairs.add(pair);

        calculateStartEnd(endPosition + 1, size);

    }

    public void shutdown() {
        try {
            this.rAccessFile.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
        this.executorService.shutdown();
    }

    private void handle(byte[] bytes) throws UnsupportedEncodingException {
        String line = null;
        if (this.charset == null) {
            line = new String(bytes);
        } else {
            line = new String(bytes, charset);
        }
        if (line != null && !"".equals(line)) {
            this.handle.handle(line);
            counter.incrementAndGet();
        }
    }


    private static class StartEndPair {
        public long start;
        public long end;

        @Override
        public String toString() {
            return "star=" + start + ";end=" + end;
        }
    }

    private class SliceReaderTask implements Runnable {
        private long start;
        private long sliceSize;
        private byte[] readBuff;

        public SliceReaderTask(StartEndPair pair) {
            this.start = pair.start;
            this.sliceSize = pair.end - pair.start + 1;
            this.readBuff = new byte[bufferSize];
        }


        @Override
        public void run() {
            try {
                MappedByteBuffer mapBuffer = rAccessFile.getChannel().map(FileChannel.MapMode.READ_ONLY, start, this.sliceSize);
                ByteArrayOutputStream bos = new ByteArrayOutputStream();
                for (int offset = 0; offset < sliceSize; offset += bufferSize) {
                    int readLength;
                    if (offset + bufferSize <= sliceSize) {
                        readLength = bufferSize;
                    } else {
                        readLength = (int) (sliceSize - offset);
                    }
                    mapBuffer.get(readBuff, 0, readLength);
                    for (int i = 0; i < readLength; i++) {
                        byte tmp = readBuff[i];
                        //碰到换行符
                        if (tmp == '\n' || tmp == '\r') {
                            handle(bos.toByteArray());
                            bos.reset();
                        } else {
                            bos.write(tmp);
                        }
                    }
                }
                if (bos.size() > 0) {
                    handle(bos.toByteArray());
                }
                cyclicBarrier.await();//测试性能用
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static class Builder {
        private int threadSize = 1;
        private Charset charset;
        private int bufferSize = 1024 * 1024;
        private IFileHandle handle;
        private File file;

        public Builder(File file, IFileHandle handle) {
            this.file = file;
            if (!this.file.exists())
                throw new IllegalArgumentException("文件不存在!");
            this.handle = handle;
        }

        public Builder threadPoolSize(int size) {
            this.threadSize = size;
            return this;
        }

        public Builder charset(Charset charset) {
            this.charset = charset;
            return this;
        }

        public Builder bufferSize(int bufferSize) {
            this.bufferSize = bufferSize;
            return this;
        }

        public BigFileReader build() {
            return new BigFileReader(this.file, this.handle, this.charset, this.bufferSize, this.threadSize);
        }
    }

}

public interface IFileHandle {
    void handle(String line);
}
0
0
分享到:
评论

相关推荐

    java nio 包读取超大数据文件

    ### Java NIO 处理超大数据文件的知识点详解 #### 一、Java NIO简介 Java NIO(New IO)是Java平台上的新输入/输出流API,它提供了与传统IO(即Java IO)不同的数据处理方式。NIO在Java 1.4版本引入,并在后续版本...

    java 读取写入csv

    在日常的软件开发中,处理CSV(Comma Separated Values)文件是一种常见的需求,特别是在数据分析、数据导入导出等场景下。Java作为一种广泛使用的编程语言,提供了多种方式来读取和写入CSV文件。本文将深入探讨如何...

    java NIO和java并发编程的书籍

    java NIO和java并发编程的书籍java NIO和java并发编程的书籍java NIO和java并发编程的书籍java NIO和java并发编程的书籍java NIO和java并发编程的书籍java NIO和java并发编程的书籍java NIO和java并发编程的书籍java...

    java NIO实例

    `NIOServer.java`和`NIOClient.java`这两个文件很可能是用于演示Java NIO服务器端和客户端的基本操作。下面将详细介绍Java NIO的主要组件和工作原理,并结合这两个文件名推测它们可能包含的内容。 1. **Selector...

    java读写csv文件,中文乱码问题

    - 使用`java.nio.charset.CharsetDetector`类检测文件的实际编码。 7. **总结**: - 解决Java处理CSV文件时的中文乱码问题,关键在于正确指定字符编码。 - 要注意文件的读取、写入以及处理过程中的编码一致性。 ...

    JavaNIO chm帮助文档

    Java NIO系列教程(一) Java NIO 概述 Java NIO系列教程(二) Channel Java NIO系列教程(三) Buffer Java NIO系列教程(四) Scatter/Gather Java NIO系列教程(五) 通道之间的数据传输 Java NIO系列教程(六)...

    java nio 读文件

    在本篇文章中,我们将深入探讨Java NIO如何读取文件。 一、NIO的基本概念 1. 缓冲区(Buffer):NIO的核心组件,用于存储数据。Java提供了多种Buffer类,如ByteBuffer、CharBuffer、IntBuffer等,分别对应不同数据...

    java NIO.zip

    Java NIO支持多种类型的通道,包括文件通道(FileChannel)、套接字通道(SocketChannel)和服务器套接字通道(ServerSocketChannel)等。通道可以同时进行读写操作,并且可以实现异步读写。 2. **缓冲区(Buffers...

    java 解析csv文件例子,csv文件 中文乱码问题

    首先,我们需要了解如何使用Java的标准库`java.io`和`java.nio`来读取CSV文件。最常用的是使用`BufferedReader`配合`Scanner`或者`Files.lines()`方法。例如: ```java import java.io.BufferedReader; import java...

    JAVA NIO 按行读取大文件,支持 GB级别

    本类,是专门为了处理大文件,按行读取开发的类。 采用读文件的缓存 fbb 1024*5 行缓存 bb 256 字节 设计思想: 每次通过nio读取字节到 fbb中 然后对fbb自己中的内容进行行判断即 10 回车 13 行号 0 文件...

    Java NIO英文高清原版

    Java NIO,全称为Non-Blocking Input/Output(非阻塞输入/输出),是Java平台中用于替代标准I/O(BIO)模型的一种新机制。NIO在Java 1.4版本引入,提供了更高效的数据处理和通道通信方式,特别适用于高并发、大数据...

    JAVA NIO 简单PFT 文件服务

    PFT(可能指的是文件传输或处理服务)在本场景中可能是通过JAVA NIO实现的一个功能模块,用于提供文件的上传、下载和列表展示服务。 文件服务是任何应用程序中不可或缺的部分,尤其是在Web环境中。在JAVA NIO中,...

    JAVA NIO 按行读取大文件支持 GB级别-修正版

    本类,是专门为了处理大文件,按行读取开发的类。 采用读文件的缓存 fbb 1024*5 行缓存 bb 256 字节 设计思想: 每次通过nio读取字节到 fbb中 然后对fbb自己中的内容进行行判断即 10 回车 13 行号 0 文件结束 ...

    nio.rar_FastCopyFile.java_NIO_UseFloatBuffer.java_java nio_文件锁

    在给定的压缩包文件中,我们关注的是"FastCopyFile.java"、"UseFloatBuffer.java"以及NIO中的文件锁功能。 首先,让我们详细了解一下`FastCopyFile.java`。这个文件很可能是一个示例程序,演示了如何使用Java NIO...

    java nio 写文件

    本篇将详细探讨Java NIO在写文件方面的应用。 1. **通道(Channels)** 在Java NIO中,数据的读取和写入都是通过通道(Channel)进行的。通道可以理解为连接到I/O设备的桥梁,例如文件、网络套接字等。要使用NIO写...

    Large-File-Processing-master_javanio_java大文件处理_

    综上所述,“Large-File-Processing-master_javanio_java大文件处理_”项目涵盖了Java NIO在大文件处理中的核心技术和最佳实践,是学习和研究Java高效处理大文件的宝贵资源。通过深入理解这些知识点,并结合项目中的...

    读取创建CSV文件并自动解析文件编码方式

    在Java中,`java.nio.charset.Charset`类用于表示字符编码,但默认情况下,Java文件读取操作通常假设文件为UTF-8编码。对于其他编码,我们需要借助第三方库如`jcconf`或`ICU4J`来检测编码。 ```java import ...

    java NIO 写文件

    在这个主题中,我们将深入探讨Java NIO如何用于写文件,特别是在处理大数据文件时如何提高性能。 首先,理解Java NIO的基本概念非常重要。NIO中的“N”代表“非阻塞”,这意味着在进行I/O操作时,程序可以继续执行...

    Java NIO 中文 Java NIO 中文 Java NIO 中文文档

    Java NIO 深入探讨了 1.4 版的 I/O 新特性,并告诉您如何使用这些特性来极大地提升您所写的 Java 代码的执行效率。这本小册子就程序员所面临的有代表性的 I/O 问题作了详尽阐述,并讲解了 如何才能充分利用新的 I/O ...

    java NIO 视频教程

    Java NIO(New IO)是一个可以替代标准Java IO API的IO API(从Java 1.4开始),Java NIO提供了与标准IO不同的IO工作方式。 Java NIO: Channels and Buffers(通道和缓冲区) 标准的IO基于字节流和字符流进行操作的,...

Global site tag (gtag.js) - Google Analytics