JAVA 并发编程-应用篇

提到java多线程不免有些人会头大，很多概念都是很理解但是真正到了实战的时候又是不知道如何操作了，下面就结合实际项目来说说多线程的应用。

业务需求：

举例：批量插入10万条用户的相关活动优惠券

操作方式：使用固定10个大小的线程池来做，并每次处理1000条插入数据

线程类：注实现Callable<Integer>接口的是能得到返回值的线程类

public class InsertBatchThread implements Callable<Integer> {

    private int vdate;
    private int uid;
    private int count;
    private FundsInfoMapper fundsInfoMapper;
    private WmpsDayInterMapper wmpsDayInterMapper;
    private DataSource dataSource;

    public WmpsDayInterMapper getWmpsDayInterMapper() {
        if (null == wmpsDayInterMapper) {
            synchronized (this) {
                if (null == wmpsDayInterMapper) {
                    wmpsDayInterMapper = SpringContextUtils.getBean("wmpsDayInterMapper");
                }
            }
        }
        return wmpsDayInterMapper;
    }

    public FundsInfoMapper getProCommFundsInfoMapper() {
        if (null == fundsInfoMapper) {
            synchronized (this) {
                if (null == fundsInfoMapper) {
                    fundsInfoMapper = SpringContextUtils.getBean("fundsInfoMapper");
                }
            }
        }
        return fundsInfoMapper;
    }

    
    /**
     * 无参构造函数
     */
    public InsertBatchThread(){

    }

    /**
     * 无参构造函数
     */
    public InsertBatchThread(int vdate,int uid,int count){
        this.vdate=vdate;
        this.uid=uid;
        this.count=count;
        this.fundsInfoMapper=getFundsInfoMapper();
        this.wmpsDayInterMapper=getWmpsDayInterMapper();
        
    }

    /**
     * 多线程规定好的方法，如果是继承Thread类则必须实现这个方法
     */
    public Integer call() {
        int result = -1;
        try {

            //操作
            List<Integer> listUsers = fundsInfoMapper.selectUserForInsertBatch(count * 1000, 1000);
            
           //批量插入用户活动优惠券记录表
            List<WmpsDayInter> listOnePageDayInner = wmpsDayInterMapper.selectByInsertBatch(vdate,listUsers.get(0),listUsers.get(listUsers.size() - 1));
            wmpsDayInterInsertBatch(listOnePageDayInner);

        }catch (Exception e){
            result=count;
        }
        return  result;
    }

    //批量插入用户活动优惠券记录表--JDBC
    public int[] wmpsDayInterInsertBatch(List<WmpsDayInter> listOnePage) throws Exception{
        dataSource=  SpringContextUtils.getBean("dataSource");
        PreparedStatement ps = dataSource.getConnection().prepareStatement("INSERT ignore INTO t_a (uid, inter, cdate) values(?,?,?)" );
        for(WmpsDayInter oneObj :listOnePage ){
            ps.setInt(1,oneObj.getUid());
            ps.setBigDecimal(2, oneObj.getInter());
            ps.setBigDecimal(3,oneObj.getCdate());
            ps.addBatch();
        }
        return ps.executeBatch();
    }

    
}

对应的业务实现类：

/**
 * 活动优惠券
 */
@Service("FundsInfoService")
public class FundsInfoServiceImpl extends  BaseService<ProCommFundsInfo, FundsInfoExample> implements IFundsInfoService {
    private final static Logger LOGGER = LoggerFactory.getLogger(FundsInfoServiceImpl.class);
    @Autowired
    private FundsInfoMapper fundsInfoMapper;

    public FundsInfoServiceImpl() {
    }

    @Override
    public void insertDayInter(Integer vdate, Integer uid) {

		//活动优惠券条件判断
        if (vdate == null || vdate.intValue() <= 0 || uid == null || uid.intValue() < 0) {
            LOGGER.error("=insertDayInter=>error,vdate=" + vdate + ",uid=" + uid);
        } else {

			//统计符合条件用户
            int totalNum = fundsInfoMapper.countForInsertBatchByUser();
            List<Integer> listException = new ArrayList<Integer>();

			//初始化第一批次起始值
			int startRow = 0;
            try {

				//固定大小线程池
                ExecutorService fixedThreadPool = Executors.newFixedThreadPool(10);

                for (int i = 0; i <= Math.floor(totalNum / 1000); i++) {
                    Future<Integer> future = fixedThreadPool.submit(new InsertBatchThread(vdate, uid, i));
                    if( future.get()>=0 ){
                        listException.add( future.get());
                    }
                }

				//活动优惠券发完，验证是否有遗漏，如果有则修复
                if(listException.size()>0){

                    for (int i = 0; i <= listException.size(); i++) {
                        Future<Integer> future = fixedThreadPool.submit(new InsertBatchThread(vdate, uid, listException.get(i)));
                        if( future.get()>=0 ){
                            listException.add( future.get());
                        }
                    }
                }

            }catch (Exception e){
                LOGGER.error("<=insertDayInter=>error",e);
            }

        }
        LOGGER.info("<=insertDayInter=>");

    }

}

问题在于，当我们需要使用多线程操作时，一般会先查询，再进行插入，但是此时如果我们没有一个合理的维度去分割数据的话，很容易造成死锁。例如如果我们没有将维度分配合理，就有可能批次一与批次二同时处理相同的数据，既要查询又要更新，互相均需要对方的锁释放才可以继续执行，导致死锁。