利用多线程实现对csv文件的处理。_多线程能同时处理一个csv文件吗_xiaolege_的博客

相关文章推荐

干练的馒头 · Java处理Exception无法捕获的异常 ...· 1 月前 ·

温文尔雅的鸵鸟 · C++ 内存分配可视化_内存可视化· 1 月前 ·

帅呆的佛珠 · Jmeter之读取csv文件踩坑记 - 梁剑lj · 1 月前 ·

绅士的机器人 · 深刻把握经济高质量发展的内涵要义--理论-中 ...· 1 年前 ·

玉树临风的柚子 · ORACLE LIKE ...· 1 年前 ·

拉风的大熊猫 · 我省推动二手车出口健康有序发展_安徽省商务厅· 1 年前 ·

爱笑的蜡烛 · 李嘉欣知三当三的后续瓜 - 知乎· 1 年前 ·

逼格高的路灯 · 家电购买不踩坑，有哪些看完可以闭眼入的选购攻 ...· 1 年前 ·

数据组同事需要一个小工具帮组他们处理一些数据。首先他们的原数据是csv文件的，需要我们把里面的数据和别的系统里面的数据逐条匹配完善后返回给他们一个csv文件。原数据的csv一般3-5万条数据量(单列的)。

解决方案：

首先读取csv文件，生成待处理数据集合。 然后把该数据集合分片，以使用多线程进行匹配处理 。最后通过POI生成csv输出。(一开始同事没有使用多线程，虽然数据量不算大，但是匹配和处理过程需要多次调用别的系统的接口服务，所以速度上还是有点慢的。他问我如何优化，这才引入多线程，处理数据速度明显提升。)

具体实现：

0、方法主干：

FileInputStream inStream = new FileInputStream(filename);
            List<String> csvList = readCsv(inStream);//--对应下面第一步
            //保证线程安全，数据不会丢失
            List<Map<String, Object>> writeCsv = Collections.synchronizedList(new ArrayList());
            List<List<String>> thredList = new ArrayList<>();
            //List分片--偏移量可以根据自己的需要来处理，这里示例代码暂且给10
            thredList = averageAssign(csvList,10);//--对应下面第二步
            //创建线程池
            ExecutorService fixedThreadPool = Executors.newFixedThreadPool(thredList.size());
            for (int i = 0; i < thredList.size(); i++) {//根据分割的list数量创建多个线程
                List<String> list = thredList.get(i);
                MyThread myThread = new MyThread(i, writeCsv, list, tokenUrl, token);//具体的数据处理逻辑都在线程的run方法中。
                Thread tempThread = new Thread(myThread);
                tempThread.setName("线程" + i);
                fixedThreadPool.execute(tempThread);
            fixedThreadPool.shutdown();//关闭线程池--当该线程池中所有线程都结束后关闭线程池
            while (true) {//等待所有任务都执行结束
                if (fixedThreadPool.isTerminated()) {//所有的子线程都结束了
                    System.out.println("共耗时:"+(System.currentTimeMillis()-startTime)/1000.0+"s");
                    break;
            try {
                long endTime1 = System.currentTimeMillis();
                float endTimePp = (float) (endTime1 - startTime) / 1000;
                logger.info("匹配结束时间：" + endTimePp + "s");
                //创建临时csv文件
                long startTim = System.currentTimeMillis();
                logger.info("文件写入开始时间：" + startTim);
                logger.info("文件写入开始时间writeCsv==：" + writeCsv.size());
                CsvUtil.createTempFile(writeCsv, urlCharset);//通过POI工具类创建csv文件并写入数据
                long endTime = System.currentTimeMillis();
                float excTime = (float) (endTime - startTim) / 1000;
                logger.info("文件写入结束时间：" + excTime + "s");
            } catch (IOException e) {
                e.printStackTrace();
                System.out.println("导出失败");
1、首先读取csv文件。 
public static List<String> readCsv(InputStream inStream) throws IOException {
        List<String> medicalWhiteListVOs = new ArrayList<String>();
        try {
            BufferedReader reader = new BufferedReader(new InputStreamReader(inStream, "UTF-8"));
            reader.readLine();//第一行信息，为标题信息，不用,如果需要，注释掉
            String line = null;
            int num = 0;
            while ((line = reader.readLine()) != null) {
                num++;
                String item[] = line.split(",");//CSV格式文件为逗号分隔符文件，这里根据逗号切分
                medicalWhiteListVOs.add(getValue(item, 0));
            logger.info("**一共行数**:" + num);
        } catch (Exception e) {
            e.printStackTrace();
        return medicalWhiteListVOs;
2、把上面获得的数据list分片，切割成多个list。 
     * 将一个list均分成n个list,主要通过偏移量来实现的
     * @param source 原list
     * @param n  要分成几个list
     * @return
    public static <T> List<List<T>> averageAssign(List<T> source,int n){
        List<List<T>> result=new ArrayList<List<T>>();
        int remaider=source.size()%n; //(先计算出余数)
        int number=source.size()/n; //然后是商
        int offset=0;//偏移量
        for(int i=0;i<n;i++){
            List<T> value=null;
            if(remaider>0){
                value=source.subList(i*number+offset, (i+1)*number+offset+1);
                remaider--;
                offset++;
            }else{
                value=source.subList(i*number+offset, (i+1)*number+offset);
            result.add(value);
        return result;
3、创建线程池，并循环创建多线程。多个线程一起处理分割后的数据集合，然后 写入同一个大的list。//创建大的数据集合用于接收各个线程处理后的数据，由于ArrayList是线程不安全的所以为了保证线程安全，数据不会丢失，在此对ArrayList进行序列化，不然会有数据丢失。
             
 List<Map<String, Object>> writeCsv = Collections.synchronizedList(new ArrayList());
//创建线程池
            ExecutorService fixedThreadPool = Executors.newFixedThreadPool(thredList.size());
            for (int i = 0; i < thredList.size(); i++) {
                List<String> list = thredList.get(i);
                MyThread myThread = new MyThread(i, writeCsv, list, tokenUrl, token);//具体处理逻辑都在线程的run方法中
                Thread tempThread = new Thread(myThread);
                tempThread.setName("线程" + i);
                fixedThreadPool.execute(tempThread);
            fixedThreadPool.shutdown();
            while (true) {//等待所有任务都执行结束
                if (fixedThreadPool.isTerminated()) {//所有的子线程都结束了
                    System.out.println("共耗时:"+(System.currentTimeMillis()-startTime)/1000.0+"s");
                    break;
4、线程的创建。 
     * 自定义线程
    class MyThread implements Runnable {
        private int i; // 第几个线程
        private List<String> csvList;
        private String token;
        private String tokenUrl;
        private List<Map<String, Object>> writeCsv;
//自定义构造方法
        public MyThread( int i,List<Map<String, Object>> writeCsv,List<String> csvList, String tokenUrl, String token) {
            this.i = i;
            this.writeCsv = writeCsv;
            this.csvList = csvList;
            this.token = token;
            this.tokenUrl = tokenUrl;
        public void run() {
            try {
                System.out.println("=====================进入线程得方法================" +Thread.currentThread().getName() + ": " + i +"--------------"+csvList.size());
                for (int j = 0; j < csvList.size(); j++) {
                    String str= csvList.get(j);//获取待处理数据
                    。。。。。。具体的处理实现需要根据自己的业务需求来编码
                    writeCsv.add(map);
            }catch (Exception e) {
                e.printStackTrace();
                System.out.println("导出失败");
5、POI写入csv。 
     * 创建临时的csv文件
     * @return
     * @throws IOException
    public static File createTempFile(List<Map<String, Object>> datas,String urlCharet) throws IOException {
        File tempFile = File.createTempFile("vehicle", ".csv");
        System.out.println("缓存路径=========="+tempFile.getCanonicalPath());
        CsvWriter csvWriter = new CsvWriter(tempFile.getCanonicalPath(), ',', Charset.forName(urlCharet));
        // 写表头
        String[] headers = {"结果地址对应的标准地址编码", "结果地址","地址层级","地址结果评分","地址经纬度","地址空间面信息","省","市","区","街道","社区"
                             ,"街路巷","门牌","小区","建筑物","单元","楼层","户室","匹配结果","原始数据","相似度评分"};
        csvWriter.writeRecord(headers);
        for (Map<String ,Object> data : datas) {
            //这里如果数据不是String类型，请进行转换
            csvWriter.write(data.get("code")==null?"":data.get("code")+"");
            csvWriter.write(data.get("addr")==null?"":data.get("addr")+"");
            csvWriter.write(data.get("lv")==null?"":data.get("lv")+"");
            csvWriter.write(data.get("score")==null?"":data.get("score")+"");
            csvWriter.write(data.get("loc")==null?"":data.get("loc")+"");
            csvWriter.write(data.get("shape")==null?"":data.get("shape")+"");
            csvWriter.write(data.get("province")==null?"":data.get("province")+"");
            csvWriter.write(data.get("city")==null?"":data.get("city")+"");
            csvWriter.write(data.get("district")==null?"":data.get("district")+"");
            csvWriter.write(data.get("street")==null?"":data.get("street")+"");
            csvWriter.write(data.get("community")==null?"":data.get("community")+"");
            csvWriter.write(data.get("road")==null?"":data.get("road")+"");
            csvWriter.write(data.get("road_num")==null?"":data.get("road_num")+"");
            csvWriter.write(data.get("village")==null?"":data.get("village")+"");
            csvWriter.write(data.get("building")==null?"":data.get("building")+"");
            csvWriter.write(data.get("building_num")==null?"":data.get("building_num")+"");
            csvWriter.write(data.get("room_floor")==null?"":data.get("room_floor")+"");
            csvWriter.write(data.get("house_num")==null?"":data.get("house_num")+"");
            csvWriter.write(data.get("ppqk")==null?"":data.get("ppqk")+"");
            csvWriter.write(data.get("yssj")==null?"":data.get("yssj")+"");
            csvWriter.write(data.get("pf")==null?"":data.get("pf")+"");
            csvWriter.endRecord();
            csvWriter.flush();
        csvWriter.close();
        return tempFile;
如果有需要输出exl表格的同学可参见以下链接：java+poi生成excel并通过浏览器实现下载_xiaolege_的博客-CSDN博客_poi导出excel浏览器下载 
及表格样式设置：POI设置excel样式_xiaolege_的博客-CSDN博客 
至此整个小工具的功能算是完成了。处理数据速度相对原来单线程提升了好几倍，3万数据只要半分钟左右。 
工具源码：demoCsv.zip-Java代码类资源-CSDN下载
                    背景：数据组同事需要一个小工具帮组他们处理一些数据。首先他们的原数据是csv文件的，需要我们把里面的数据和别的系统里面的数据逐条匹配完善后返回给他们一个csv文件。原数据的csv一般3-5万条数据量(单列的)。解决方案：首先读取csv文件，生成待处理数据集合。然后把该数据集合分片，以使用多线程进行匹配处理。最后通过POI生成csv输出。(一开始同事没有使用多线程，虽然数据量不算大，但是...
				作为Go脚本的一部分，我一直在努力，我想从多个Go例程中写入CSV文件，但是意识到内置的CSV Writer不是线程安全的。 
 我第一次写入CSV文件的尝试如下所示： 
package main
import (
	"encoding/csv"
	"log"
	"strconv"
func main() {
	csvFile, err := os.Creat...
				Python如果单线程执行代码去处理数万个xlsx，包括读出和写入的操作，整个过程耗时会很长。本文以处理一批15000个csv文件为例，对比Python3单线程和多线程处理效率。
任务：每一个csv包括三个波段的内容，将其分解成三个波段，总共产生45000个文件。
import os
from pandas import Series
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import re
import
				领导说公司的报表导出太慢还没有进度条展示,以前用的poi操作excel导出xls,从数据库查询数据,写入本地文件,然后读取本地文件下载效率低下,特别是数据量大的时候。所以我对后台cms系统导出做了优化，采用导出csv格式，这种格式下速度比xls快很多。
**csv和xls的区别**
   经常会用户报表导出格式为csv和xls格式。他们的性能相差很大，我认为csv是非常高效的格式，而xls性能虽...
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.io.Buf