多线程优化

   public void newExecute() {
        //获取线程数，默认30
        int threadNum = 30;
        String strThreadNum = LionConfigUtils.getProperty("ts-monitor-job.dailyJob.accountBalanceDailyCheckerThreadNum",
                "");
        if (isNumeric(strThreadNum)) {
            threadNum = Integer.parseInt(strThreadNum);
        }
        ExecutorService service = Executors.newFixedThreadPool(threadNum);
        //所有账号总数
        int accountCount = accountDao.findAllMerchantAccountCount();
        int latchCount = accountCount % BATCH_SIZE == 0 ? accountCount / BATCH_SIZE : (accountCount / BATCH_SIZE + 1);
        final CountDownLatch latch = new CountDownLatch(latchCount);
        final Date bizDate = DateUtils.addDate(DateUtils.removeTime(new Date()), -1);
        final Date lastBizDate = DateUtils.addDate(bizDate, -1);
        int offset = 0;
        while (offset < accountCount) {
            offset += BATCH_SIZE;
            final int initPage = offset;
            Runnable run = new Runnable() {
                public void run() {
                    try {
                        //todo
                        List<AccountData> accountDataList = accountDao.findAllMerchantAccount(initPage, BATCH_SIZE);
                        for (AccountData accountData : accountDataList) {
                            Thread.sleep(20);
                            doAccountMonitor(bizDate, lastBizDate, accountData);
                        }
                    } catch (Exception e) {
                        monitorLogger.error(String.format("severity=[1], query account data list fail! offset = " +
                                "[%d]...", initPage), e);
                    } finally {
                        latch.countDown();
                    }
                }
            };
            service.submit(run);
        }

        try {
            latch.await();
        } catch (InterruptedException e) {
            monitorLogger.error("CountDownLatch.await() error:" + e);
        }
    }

 上面的代码没起到优化作用，下面的代码优化比较好

public String execute() throws Exception {

		String timeFlag=LionConfigUtils.getProperty("ts-tg-settle-service.tgSettleGroupJob.timeFlag");
		int  sleepTime= Integer.parseInt(LionConfigUtils.getProperty("ts-tg-settle-service.tgSettleGroupJob.sleepTime"));
		//生成查询的初始时间和结束时间
		String startTime=null;
		String endTime=null;
		String timelist[]=	generateTargetDate(timeFlag);
		startTime=timelist[0];
		endTime=timelist[1];
		//把要查询的状态封装入list 默认只查init类型
		List<Integer> statusList=new ArrayList<Integer>();
		statusList.add(GroupSettleStatus.INIT.getCode());
		//便利group表,一边遍历一边发消息
		//todo 分页 dao写sql 每页大小lion配置

		//分页数据 startRow endRow pageSize 注意，pagesize从lion获取，检验一下是否为数字
		int startRow=0;

		int pageSize=10;
		//线程数默认30个先
		int threadNum=30;

		String  strPageSize= LionConfigUtils.getProperty("ts-tg-settle-service.tgSettleGroupJob.pageSize");

		if(isNumeric(strPageSize)){
			pageSize=Integer.parseInt(strPageSize);
		}
		String strThreadNum= LionConfigUtils.getProperty("ts-tg-settle-service.tgSettleGroupJob.groupThreadNum");
		if(isNumeric(strThreadNum)){
			threadNum=Integer.parseInt(strThreadNum);
		}


		ExecutorService service = Executors.newFixedThreadPool(threadNum);

		//得到所有要发的数量和latch数量。
		int resultCount=tstgSettleGroupDao.selectCountSettleGroupToTuanGouJob(startTime,endTime,statusList);
		int latchCount=0;
		if(resultCount%pageSize==0){
			latchCount=resultCount/pageSize;
		}else{
			latchCount=resultCount/pageSize+1;
		}
		CountDownLatch latch=new CountDownLatch(latchCount);

		//查询每页数据并发送 直到查没有了
        int lastId = 0;
        List<TSTGSettleGroupEntity> entityList = null;
		while(true){
            if(lastId == 0){
                entityList = tstgSettleGroupDao.selectSettleGroupToTuanGouJobByAddTime(startRow,pageSize,startTime,endTime,statusList,lastId);
            }else{
                entityList=tstgSettleGroupDao.selectSettleGroupToTuanGouJob(startRow,pageSize,startTime,endTime,statusList,lastId);
            }

			if(CollectionUtils.isEmpty(entityList)){
				break;
			}
			//logger.info(times+" send group ： pageSize: "+pageSize +"  startRow:"+startRow +"  startTime: "+startTime+"  endTime: "+endTime);
			//System.out.println(lastId+" send group ： pageSize: "+pageSize +"  startRow:"+startRow +"  startTime: "+startTime+"  endTime: "+endTime);
            service.submit(new ManageListAndSendThread(tstgSettleGroupDao,entityList,settleAccountMessageProducer,latch,sleepTime));
            lastId = entityList.get(entityList.size() - 1).getId();
		}
        try {
            latch.await();
        }catch (Exception e){
            logger.error("latch await error!"+e.getMessage());
        }

		return null;
	}

 

public class ManageListAndSendThread implements Runnable{
///.............省略部分代码
    @Override
    public void run() {
        try{
            for(TSTGSettleGroupEntity entity:entityList){

                TuanGouSettleAccountMessageDTO messageDTO=transformSettleEntityToMessageDTO(entity, ResourceType.GROUP.getIdentifier());

                logger.info(" i send outBizId :"+messageDTO.getOutBizId()+" threadName："+Thread.currentThread().getName());

                    //大量数据下出现个案 所以逻辑改为先修改状态再发送
                    //把entity 状态改为发送中
                    TSTGSettleGroupEntity temp=new TSTGSettleGroupEntity();
                    temp.setId(entity.getId());
                    temp.setSettleStatus(GroupSettleStatus.SETTLING.getCode());
                    tstgSettleGroupDao.update(temp);
                    logger.info("状态修改为已发送，groupId为："+entity.getId());
                    temp=null;
                    sendTgGroupSettle(messageDTO);
                    logger.info("消息已发送，groupId为："+entity.getId());

                try {
                    Thread.sleep(sleepTime);
                }catch (Exception e){

                }
            }
        }finally {
            //计数器减一
            latch.countDown();
        }
    }
}

 

这段代码结合多线程对大数据量下的分页查询进行了优化，SQL语句参考下面

SELECT ID,AccountID,AccountType,AccountSubject,Credit,Debit,STATUS,IsBuffer,AccountDirection,Balance,FrozenBalance,ADDTIME,UpdateTime
        FROM Account
        WHERE AccountType = 2
        ORDER BY ID
        LIMIT 1825000, 10;
SELECT ID,AccountID,AccountType,AccountSubject,Credit,Debit,STATUS,IsBuffer,AccountDirection,Balance,FrozenBalance,ADDTIME,UpdateTime
        FROM Account
        WHERE AccountType = 2 AND id >= 2790048
        ORDER BY ID ASC
        LIMIT 10;
SELECT ID,AccountID,AccountType,AccountSubject,Credit,Debit,STATUS,IsBuffer,AccountDirection,Balance,FrozenBalance,ADDTIME,UpdateTime
        FROM Account 
        WHERE AccountType = 2 AND id >(
		SELECT id FROM  Account ORDER BY id LIMIT 1825000,1       
        ) ORDER BY id LIMIT 10;