多线程批量插入数据库

背景

目前有12万数据需要打入到mysql数据库

准备

开发环境 ：Java、JdbcTemplate、ThreadPoolTaskExecutor（线程池）

数据库：MySql

代码：

1、配置线程池

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
@EnableAsync
public class ExecutorConfig {
    @Bean
    public Executor asyncServiceExecutor(){
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(5);
        //配置最大线程数
        executor.setMaxPoolSize(10);
        //配置队列大小
        executor.setQueueCapacity(400);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix("thread-");
        // rejection-policy：当pool已经达到max size的时候，如何处理新任务
        // CALLER_RUNS：不在新线程中执行任务，而是有调用者所在的线程来执行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化
        executor.initialize();
        return executor;
    }
}

2、Controller层分线程执行，执行多线程插入数据库

import com.example.demo_project.model.XmlEnglishBook;
import com.example.demo_project.service.XmlEnglishBookService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

@RestController
@RequestMapping(value = "xmlEnglishBook")
public class XmlEnglishBookController {

    private CountDownLatch countDownLatch;

    @Autowired
    private XmlEnglishBookService xmlEnglishBookService;

    @RequestMapping(value = "test",method = RequestMethod.GET)
    public String test(){
        return "这是一个测试接口";
    }


    @RequestMapping(value = "readXml2Database",method = RequestMethod.GET)
    public String readXml2Database() throws Exception{
        // 在这里 list 需要复制 需要插入数据库的实例集合（例如12万实体集合）
        List list = new ArrayList<>();

        // 异步多线程  插入数据库
        int result = 0;
        int total = list.size();
        int batchSize = 2000;
        int number =total%batchSize ==0 ?total / batchSize :total / batchSize+1;
        countDownLatch = new CountDownLatch(number);
        for(int i = 0;i batchList = new ArrayList<>();
            if(i== number-1){  // 最后一个
                batchList = list.subList(i*batchSize,total);
            }else{
                batchList = list.subList(i*batchSize,(i+1)*batchSize);
            }
            xmlEnglishBookService.asyncBatchAddXmlEnglishBook(countDownLatch,batchList);
        }

//        return xmlEnglishBookService.batchAddXmlEnglishBook(list);
        return "结果-"+result;
    }
}

解释：// 在这里 list 需要复制 需要插入数据库的实例集合（例如12万实体集合）
        List list = new ArrayList<>(); 
XmlEnglishBook：是自己的数据库实体
3、service 接口层和实现层

import java.util.List;
import java.util.concurrent.CountDownLatch;

public interface XmlEnglishBookService {

    Integer asyncBatchAddXmlEnglishBook(CountDownLatch countDownLatch, List batchList);
}

import com.example.demo_project.dao.IXmlEnglishBookDao;
import com.example.demo_project.model.XmlEnglishBook;
import com.example.demo_project.service.XmlEnglishBookService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.util.List;
import java.util.concurrent.CountDownLatch;

@Service
public class XmlEnglishBookServiceImpl implements XmlEnglishBookService {

    @Autowired
    private IXmlEnglishBookDao xmlEnglishBookDao;


    @Override
    @Async("asyncServiceExecutor")
    public Integer asyncBatchAddXmlEnglishBook(CountDownLatch countDownLatch, List batchList) {
        try {
            int count = xmlEnglishBookDao.batchAdd(batchList);
            if(count>0){
                return Integer.valueOf(count);
            }else{
                return 0;
            }
        } catch (Exception e) {
            System.out.println(e.getLocalizedMessage());
            return 0;
        }finally {
            countDownLatch.countDown();
        }
    }
}

注意：实现插入数据的接口需要异步@Async("asyncServiceExecutor")

4、dao层接口与实现

import com.example.demo_project.model.XmlEnglishBook;

import java.util.List;

public interface IXmlEnglishBookDao {

    int batchAdd(List list);
}

import com.example.demo_project.dao.IXmlEnglishBookDao;
import com.example.demo_project.model.XmlEnglishBook;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.jdbc.core.BeanPropertyRowMapper;
import org.springframework.jdbc.core.namedparam.BeanPropertySqlParameterSource;
import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
import org.springframework.jdbc.core.namedparam.SqlParameterSource;
import org.springframework.jdbc.core.namedparam.SqlParameterSourceUtils;
import org.springframework.stereotype.Repository;

import java.util.List;

@Repository
public class XmlEnglishBookDao implements IXmlEnglishBookDao {

    @Autowired
    private NamedParameterJdbcTemplate jdbcTemplate;
    private String table = "表名称";


    @Override
    public int batchAdd(List list) {
        StringBuilder strSql = new StringBuilder();
        strSql.append("INSERT INTO "+table+" (");
        strSql.append("biao_zi_duan)");
        strSql.append(" VALUES(");
        strSql.append(":biao_zi_duan)");
        SqlParameterSource[] paramSourceArray = SqlParameterSourceUtils.createBatch(list);
        int[] result = jdbcTemplate.batchUpdate(strSql.toString(),paramSourceArray);
        if(result != null){
            return result.length;
        }else{
            System.out.println("-------"+list.get(0).getBook_name());
            return 0;
        }

    }
}