Welcome to MLink Developer Q&A Community for programmer and developer-Open, Learning and Share
Welcome To Ask or Share your Answers For Others

Categories

0 votes
456 views
in Technique[技术] by (71.8m points)

分页查询中如何使用多线程加快处理速度?

一个历史表中有大量的数据,现在要通过分页式查询处理转换数据。

现在将处理数据的逻辑放在线程池中处理,以加快处理流程。

可是总是出现事务方面的异常

比如 : SQLNonTransientConnectionException

请问该如何解决上述异常,或者有什么好的多线程分页查询处理方案


原来问题描述不太清楚,现在添加以下代码(手敲,如果有错,请多包含)

分页式查询逻辑:

int pageSize = 100;
int currentPageLength = 0;
int pageIndex = 0;
ExecutorService exe  = newFixedThreadPool(Runtime.getRuntime().availableProcessors());
do {
    int offset = pageIndex * pageSize;
    List<TradeInfo> tradeInfos = tradeInfoService.findTradeInfoBysPage(queryParams,offset,pageSize);
    if (null != tradeInfos && tradeInfos.size() > 0) {
        currentPageLength = tradeInfos.size();
        TradeInfoProcesserTask task = new TradeInfoProcesserTask(tradeInfos );
        exe.execute(task);
        pageIndex++;
    }else{
        System.out.println("Page Query TradeInfo Got NOTHING! Break query loop!");
        break;
    }
} while (currentPageLength == pageSize);

exe.shutdown();

while(true) {
    if(exe.isTerminated()){
        doOtherThings();
        System.out.println("分页式多线程处理数据完毕!");
        break;
    }
}

数据处理逻辑:

public class TradeInfoProcesserTask implements Runnable{
    private volatile List<TradeInfo> tradeInfos;
    
    public TradeInfoProcesserTask (List<TradeInfo> _tradeInfos){
        tradeInfos = _tradeInfos;
    }
    
    @Override
    public void run() {
        processTradeInfos();
    }
    
    private void processTradeInfos(){
        //do something with tradeInfos .....
    }
}

与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…
Welcome To Ask or Share your Answers For Others

1 Answer

0 votes
by (71.8m points)

逻辑先不说。
现在没有判断多线程是否全部执行完,while循环完就shutdown。。。
将CountDownLatch通过构造器传入线程

图片描述

ExecutorService exe  = newFixedThreadPool(Runtime.getRuntime().availableProcessors());
CountDownLatch latch = new CountDownLatch(?); //?代表开启全部线程的数量
do {
    int offset = pageIndex * pageSize;
    List<TradeInfo> tradeInfos = tradeInfoService.findTradeInfoBysPage(queryParams,offset,pageSize);
    if (null != tradeInfos && tradeInfos.size() > 0) {
        currentPageLength = tradeInfos.size();
        TradeInfoProcesserTask task = new TradeInfoProcesserTask(tradeInfos, latch);
        exe.execute(task);
        pageIndex++;
    }else{
        System.out.println("Page Query TradeInfo Got NOTHING! Break query loop!");
        break;
    }
} while (currentPageLength == pageSize);

latch.await(); //多线程全部执行完
exe.shutdown();
doOtherThings();
System.out.println("分页式多线程处理数据完毕!");

public class TradeInfoProcesserTask implements Runnable{
    private volatile List<TradeInfo> tradeInfos;
    private CountDownLatch latch;
    
    public TradeInfoProcesserTask (List<TradeInfo> _tradeInfos, CountDownLatch latch){
        tradeInfos = _tradeInfos;
        this.latch = latch;
    }
    
    @Override
    public void run() {
        processTradeInfos();
        latch.countDown();
    }
    
    private void processTradeInfos(){
        //do something with tradeInfos .....
    }
}

与恶龙缠斗过久,自身亦成为恶龙;凝视深渊过久,深渊将回以凝视…
Welcome to MLink Developer Q&A Community for programmer and developer-Open, Learning and Share
...