需求：把一个500M的txt文件导入mysql数据库，数据量大概有几千万。

1、项目架构

项目采用微服务架构，上传文件的后台管理系统是作为应用层，业务处理的作为服务层，应用层和服务层都有多个节点。

如果是单一节点处理，那么读取500M的文件(可能更大)，并把几百万的数据要快速的写入数据库，是非常有难度的。但是我们可以利用微服务多个节点分散处理，即应用层只读取数据，读10000条就放入线程池向服务层发起请求，而服务层有多个节点，可以把应用层的请求平均到多个节点处理，在应用层每个节点先做数据解析，然后插入mysql。

2、服务层处理

2.1 文件读取

文件读取我们用BufferedReader，BufferedReader使用装饰器模式，它的IO行为是每次读进来8K的数据到缓冲区(当然，缓冲区的大小我们是可以通过构造器修改的)，如果需要使用数据的时候，再直接从缓冲区里面拿出数据来使用。

而FileReader的read方法，每调用一次就会read一次file，进行一次IO。不管是多次read还是一次性的read，都不是很优雅的在read文件的方式。多次read必然会产生多次IO，一次性的read如果遇到很大的文件，对内存是极不友好的。

所以BufferedReader既能提高的读取速度，又节省了IO的次数，是一种比较优雅的读取文件的方式。
BufferedWriter和FIleWriter同理。

研究一下BufferedReader的源码，就会发现，BufferedReader中对文件的读取还是通过FileReader来实现的，BufferedReader只是对其读取到的数据做一下缓冲，api如下。其中buffer操作的api和java nio中对buffer的操作类似。

有缓冲区 VS 没有缓冲区

  • 没有缓冲区时，每次读取操作都会导致一次文件读取操作（就是告诉操作系统内核我要读这个文件的这个部分，麻烦你帮我把它取过来）。
  • 有缓冲区时，会一次性读取很多数据，然后按要求分次交给上层调用者。
    读取块大小通常是按最适合硬件的大小来读的，因为对于硬件来说，一次读取一块连续数据（比如 1K）和一次读取一个字节需要的时间几乎是一样的（都是一次读操作，只是最终提交的数据量有差异），所以带缓冲的 I/O 和不带缓冲的相比效率差异是非常显著的

我们这里的需求是顺序读取，如果是随机读取，则使用RandomAccessFile。
所谓随机读取，就是说我们需要自由访问文件的任意位置（指定位置读，指定位置写），所以如果需要访问文件的部分内容，RandomAccessFile将是更好的选择。所以当我们要下载一个大文件时，可以通过多线程使用RandomAccessFile来实现。同样的，对于文件的切割、合并，使用RandomAccessFile效率都会很高。

2.2 业务处理

BufferedReader读取文件后，每读10000行，就丢入线程池，然后调用服务层处理，应用层这里不做任何业务逻辑处理，因为应用层必然是用一个节点处理业务，但是=我们通过http调用服务层后，是可以通过多个节点处理的，这样就可以提升处理效率。


ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(10, 50, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>());
        
BufferedReader bufferedReader = null;
        try {
            bufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream("E:\\code\\esg-cemp-core-nonactivity-01394067\\logs\\system\\system2.log"), StandardCharsets.UTF_8));
       
            List<String> lines = new ArrayList<>();
            String lineTXT="";
            long count = 0;
            while ((lineTXT = bufferedReader.readLine()) != null) {
               if(count % 10000 == 0){
                   executorService.execute(new Runnable() {
                       @Override
                       public void run() {
                           //调用微服务
                           System.out.println("------调用微服务--------");
                       }
                   });
                   lines.clear();
               }
               lines.add(lineTXT);
               count++;
            }

            if(lines.size()>0){
                //调用微服务
                System.out.println("-------调用微服务-------");
            }
            //及时关闭线程池，对于这种使用不是很频繁的线程池使用完毕以后，可以及时关闭以节省资源
           //shutdown关闭以后，就不能再提交任务了，
            threadPoolExecutor.shutdown();
        } catch (Exception e) {
            e.printStackTrace();
        }

3、业务层处理

3.1 业务处理

业务处理比较简单，就是把String 切割成多个字段，然后变换成PO对象。

3.2 mysql 插入

  • 方法一：mybatis foreach
    foreach的作用是构建in条件，通过foreach可以在SQL语句中进行迭代一个集合。
    优点是使用方便；
    缺点是每条sql是有长度限制的，所以list的数量跟表字段多少直接关联；

    activityProductUserFlowPoMapper.insertList(list2);

    <insert id="insertList" >
  insert into t_non_standard_act_user_flow (act_pro_id, user_id,
    mobile, point, status,
    extend)
  values
  <foreach collection="list" separator="," item="item">
    (#{item.actProId,jdbcType=BIGINT}, #{item.userId,jdbcType=VARCHAR},
    #{item.mobile,jdbcType=VARCHAR}, #{item.point,jdbcType=INTEGER}, #{item.status,jdbcType=TINYINT},
    #{item.extend,jdbcType=VARCHAR})
  </foreach>
</insert>

  • 方法二：mybatis batch
    Mybatis内置的ExecutorType有3种，默认的是simple，该模式下它为每个语句的执行创建一个新的预处理语句，单条提交sql；而batch模式重复使用已经预处理的语句，并且批量执行所有更新语句。

    SqlSession sqlSession = sqlSessionFactory.openSession(ExecutorType.BATCH);
Mapper userMapper = sqlSession.getMapper(Mapper.class);
int size = 10000;
  try {
      for (int i = 0; i < size; i++) {
              User user = new User();
              user.setName(String.valueOf(System.currentTimeMillis()));
              fooMapper.insert(foo);
             if (i % 1000 == 0 || i == size - 1) {
                  //手动每1000个一提交，提交后无法回滚
                  session.commit();
                  //清理缓存，防止溢出
                 session.clearCache();
              }
      }
  } catch (Exception e) {
     //没有提交的数据可以回滚
    session.rollback();
  } finally {
    session.close();
  }

    需要在jdbc连接url上追加rewriteBatchedStatements=true，否则不起作用

  • 方法三：jdbc batch
    采用PreparedStatement.addBatch()方式实现，发送的是预编译后的SQL语句，执行效率高。

         Long start = System.currentTimeMillis();
      DruidDataSource dataSource = SpringUtils.getBean("dataSource");
      //从连接池中获取connection
      DruidPooledConnection conn =dataSource.getConnection();
      conn.setAutoCommit(false);
      try {
          String sql = "insert into act_user_flow (act_pro_id, user_id, mobile, point, status) values(?,?,?,?,?)";
          PreparedStatement ps = conn.prepareStatement(sql);

          for (int i = 0; i < list.size; i++) {
              ps.setLong(1,list[i].getActProId());
              ps.setString(2,list[i].getUserId());
              ps.setString(3,list[i].getMobile());
              ps.setInt(4,list[i].getPoint());
              ps.setInt(5,DuobaoGoodEnum.ACTIVITY_PRODUCT_ONLINE.getCode());
              ps.addBatch();
              //小批量提交,避免OOM
              if((i+1) % 1000 == 0) {
                  ps.executeBatch();
                  ps.clearBatch();
              }
          }
          //提交剩余的数据
          ps.executeBatch();
          conn.commit();
      } catch (SQLException throwables) {
          throwables.printStackTrace();
      } finally {
          //使用完以后放回连接池
          conn.close();
      }
      System.out.println("insert cost time = " + (System.currentTimeMillis()-start));

    需要在jdbc连接url上追加rewriteBatchedStatements=true，否则不起作用

  • 性能比较
    同个表插入一万条数据时间近似值：
    JDBC BATCH 1.1秒左右 > Mybatis BATCH 2.2秒左右 >Mybatis foreach 4.5秒左右

    方法二和方法三利用的是mysql的批量提交，需要在jdbc连接url上追加rewriteBatchedStatements=true，如下

    jdbc:mysql://xxx.com.cn/xxx?useUnicode=true&amp;characterEncoding=utf8&allowMultiQueries=true&rewriteBatchedStatements=true

    MySQL JDBC驱动在默认情况下会无视executeBatch()语句，把我们期望批量执行的一组sql语句拆散，一条一条地发给MySQL数据库，批量插入实际上是单条插入，直接造成较低的性能。
    只有把rewriteBatchedStatements参数置为true, 驱动才会帮你批量执行SQL
    另外这个选项对INSERT/UPDATE/DELETE都有效

扩展：

  • jdbc连接url上allowMultiQueries=true的作用：

1.可以在sql语句后携带分号，实现多语句执行。
2.可以执行批处理，同时发出多个SQL语句。
  • $ 和#的区别

#{}：是以预编译的形式，将参数设置到sql语句中；PreparedStatement；防止sql注入
${}：取出的值直接拼装在sql语句中；会有安全问题；
如下所示：
select * from tbl_employee where id=${id} and last_name=#{lastName}
Preparing: select * from tbl_employee where id=2 and last_name=?

如果id为 '2 or 1=1'，则会发生数据泄露，这就是sql注入。
Preparing: select * from tbl_employee where id=2 or 1=1 and last_name=?
  • 获取DruidDataSource当前活跃连接数

//当前活跃的，即正在被使用的连接数
System.out.println(dataSource.getActiveCount()); ——活跃连接都在activeConnections(Map)中

//最大连接数
System.out.println("========="+dataSource.getMaxActive()); 

//poolingCount值代表剩余可用的连接数，每次从末尾拿走连接
System.out.println(dataSource.getPoolingCount());  ——剩余可用连接都在connections(数组)中

//activeCount + poolingCount只能小于等于maxActive
activeCount + poolingCount <= maxActive



作者：张凯_9908
链接：https://www.jianshu.com/p/6b0e27e5b482
来源：简书
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

# java # mysql # 数据库
Logo
华为云开发者联盟

为开发者提供学习成长、分享交流、生态实践、资源工具等服务，帮助开发者快速成长。

更多推荐

cover

Redis开源社区持续壮大，华为云为Valkey项目注入新的活力

avatar 华为云开发者联盟
cover

Python文本统计与分析：从基础到进阶

avatar 华为云开发者联盟
cover

教你用Perl实现Smgp协议

avatar 华为云开发者联盟