Java并发任务处理之Executor线程池

叁歲伎倆 2022-04-18 05:41 252阅读 0赞

## 干货 ##

import org.junit.After;
    import org.junit.Test;
    
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    public class ThreadPoolTest { 
    
        private ExecutorService executorService;
    
        /** * 监听至所有线程结束 */
        @After
        public void terminated() { 
            executorService.shutdown();
            while(true) { 
                if(executorService.isTerminated()) { 
                    return;
                } else { 
                    try { 
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
        }
    
        /** * 缓存线程池，如果线程池长度超过处理需要，可灵活回收空闲线程，若无可回收，则新建线程。 */
        @Test
        public void testCachedThreadPool() { 
            executorService = Executors.newCachedThreadPool();
            for (int i = 0; i < 5; i++) { 
                executorService.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println("Runing...");
                    }
                });
            }
        }
    
        /** * 定长线程池，可控制线程最大并发数，超出的线程会在队列中等待。 */
        @Test
        public void testFixedThreadPool() { 
            executorService = Executors.newFixedThreadPool(10);
            for (int i = 0; i < 50; i++) { 
                executorService.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println("Runing...");
                    }
                });
            }
        }
    
        /** * 单线程化的线程池，它只会用唯一的工作线程来执行任务，保证所有任务按照指定顺序(FIFO, LIFO, 优先级)执行。 */
        @Test
        public void testSingleThreadExecutor() { 
            executorService = Executors.newSingleThreadExecutor();
            for (int i = 0; i < 50; i++) { 
                executorService.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        System.out.println("Runing...");
                    }
                });
            }
        }
    }

## 应用场景 ##

现有一业务，对数据库内用户账户信息更新，需要遍历10k+Ethereum地址，并访问geth客户端，获取地址内的余额。项目内使用QuartzJob定时执行该这一业务，代码：

public class BalanceJob extends QuartzJobBean { 
        private static final Logger logger = LoggerFactory.getLogger(BalanceJob.class);
    
        @Autowired
        private AccountMapper accountMapper;
    
        @Override
        protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException { 
            logger.info("==>Update balance {}", new Date());
            long start = System.currentTimeMillis();
            AccountExample accountExample = new AccountExample();
            AccountExample.Criteria accountExampleCriteria = accountExample.createCriteria();
            accountExampleCriteria.andIsDeleteEqualTo(false);
            List<Account> accountList = accountMapper.selectByExample(accountExample);
            List<Account> recordList = new ArrayList<>(accountList.size());
            logger.info("==>record amount {}", accountList.size());
            WalletUtil.getWeb3j();
            ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
            accountList.forEach(account -> { 
                cachedThreadPool.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        Crypto crypto = account.getCrypto();
                        Coin coin = account.getCoin();
                        if (null == crypto || null == coin) { 
                            return;
                        }
                        BigDecimal balanceOri = account.getBalance();
                        String address = crypto.getAddress();
                        String contract = coin.getContract();
                        String symbol = coin.getSymbol();
                        int decimals = coin.getDecimals();
                        BigDecimal balance = BigDecimal.ZERO;
                        if ("ETH".equalsIgnoreCase(symbol)) { 
                            try { 
                                balance = WalletUtil.getBalance(address);
                            } catch (Exception e) { 
                                e.printStackTrace();
                            }
                        } else if (StringUtils.isNotBlank(contract)) { 
                            try { 
                                balance = WalletUtil.getBalance(address, contract, decimals);
                            } catch (Exception e) { 
                                e.printStackTrace();
                            }
                        } else { 
                            logger.error("==>Coin is not exist : ", symbol);
                            return;
                        }
                        if (BigDecimal.ZERO.compareTo(balance) != 0 && balance.compareTo(balanceOri) != 0) { 
                            Account record = new Account.Builder().id(account.getId()).balance(balance).build();
                            recordList.add(record);
                        }
                    }
                });
            });
            cachedThreadPool.shutdown();
            while (true) { 
                if (cachedThreadPool.isTerminated()) { 
                    break;
                } else { 
                    try { 
                        Thread.sleep(1000L);
                    } catch (InterruptedException e) { 
                        e.printStackTrace();
                    }
                }
            }
            if (recordList.size() > 0) { 
                accountMapper.batchUpdateBalanceById(recordList);
            }
            long end = System.currentTimeMillis();
            logger.info("<==Update balance end. took {}ms", end - start);
        }
    }

使用CachedThreadPool处理getBalance方法。优势：16k条记录可以在6000ms左右处理完毕；劣势：与需求“与生俱来”，增加以太坊RPC客户端负荷。