1、Java
當(dāng)要使用線程去執(zhí)行一個(gè)任務(wù)時(shí)，可以使用ExecutorService.submit(new Callable);
這樣可以不影響其他的業(yè)務(wù)的執(zhí)行，但是在線程中的異常不能捕獲到，也不能知道線程的執(zhí)行情況；
首先創(chuàng)建一個(gè)固定大小堵塞的線程池：
public class O2oThreadPoolExecutor extends ThreadPoolExecutor {
private static final Logger logger = LoggerFactory.getLogger(O2oThreadPoolExecutor.class);
private static ExecutorService executor = new O2oThreadPoolExecutor(2, 100, 60, TimeUnit.SECONDS,
new ArrayBlockingQueue(500), new RejectedExecutionHandlerImpl());
private static int MAX_CONCURRENCY = 0;
public static ExecutorService getInstance(){
return executor;
}
public O2oThreadPoolExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit,
BlockingQueue workQueue, RejectedExecutionHandler handler) {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
}
@Override
public void shutdown() {
printThreadPoolInfo("Shutdown");
super.shutdown();
}
@Override
protected void beforeExecute(Thread t, Runnable r) {
printThreadPoolInfo("Before execute");
super.beforeExecute(t, r);
}
@Override
protected void afterExecute(Runnable r, Throwable t) {
printThreadPoolInfo("After execute");
super.afterExecute(r, t);
}
private void printThreadPoolInfo(String subject){
int runningTasks = getActiveCount();
MAX_CONCURRENCY = runningTasks > MAX_CONCURRENCY ? runningTasks : MAX_CONCURRENCY;
logger.info("{}, pool size:{}, running task:{}, pending task:{}, max concurrency:{}.",
subject,getPoolSize(),getActiveCount(),getQueue().size(),MAX_CONCURRENCY);
}
}
然后實(shí)現(xiàn)RejectedExecutionHandler，主要是當(dāng)拒絕之后，重新put進(jìn)去，防止丟失
public class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
if(!executor.isShutdown()){
try{
executor.getQueue().put(r);
}catch(InterruptedException e){
}
}
}
}
在然后定義線程需要執(zhí)行的任務(wù)，實(shí)現(xiàn)Callable接口
public class TradeStatisticsTask implements Callable {
private final static Log logger = LogFactory.getLog(TradeStatisticsTask.class);
private UserTradeStatisticsRequestDTO userTradeStatisticsRequestDTO;
private UserTradeStatisticsService userTradeStatisticsService;
private static final String TRADE_STATISTICS_CHANGE_LOCK = "trade_statistics_change_lock";
private static final int TRADE_STATISTICS_CHANGE_LOCK_EXPIRE = 1;
public TradeStatisticsTask(UserTradeStatisticsService userTradeStatisticsService,UserTradeStatisticsRequestDTO userTradeStatisticsRequestDTO){
this.userTradeStatisticsRequestDTO = userTradeStatisticsRequestDTO;
this.userTradeStatisticsService = userTradeStatisticsService;
}
@Override
public Long call() throws Exception {
logger.info("TradeStatisticsTask的insert任務(wù)："+JSONUtils.toJsonString(this.userTradeStatisticsRequestDTO));
O2oLock lock = new O2oRedisLock(TRADE_STATISTICS_CHANGE_LOCK + userTradeStatisticsRequestDTO.getBankCardNumber()+userTradeStatisticsRequestDTO.getUserId(),
TRADE_STATISTICS_CHANGE_LOCK_EXPIRE);
if (lock.tryLock(TRADE_STATISTICS_CHANGE_LOCK_EXPIRE)) {
try{
//先判斷在統(tǒng)計(jì)數(shù)據(jù)中是否有該用戶、銀行卡、當(dāng)天的記錄，如果沒有，則插入，否則更新
Long userId = this.userTradeStatisticsRequestDTO.getUserId();
String bankCardNumber = this.userTradeStatisticsRequestDTO.getBankCardNumber();
String tradeYearMouthDay = this.userTradeStatisticsRequestDTO.getTradeYearMouthDay();
UserTradeStatistics tradeParam = userTradeStatisticsService.queryUserTradeByUserIdAndBankCardNum(tradeYearMouthDay, userId, bankCardNumber);
if(tradeParam != null){
logger.info("需要更新的tradestatistics是："+JSONUtils.toJsonString(tradeParam));
BigDecimal dayAmount = this.userTradeStatisticsRequestDTO.getDayAmount().add(tradeParam.getDayAmount());
tradeParam.setDayAmount(dayAmount);
userTradeStatisticsService.update(tradeParam);
return tradeParam.getId();
}else{
logger.info("開始插入統(tǒng)計(jì)數(shù)據(jù)");
UserTradeStatistics userTradeStatistics = new UserTradeStatistics();
BeanMapper.copy(this.userTradeStatisticsRequestDTO, userTradeStatistics);
return userTradeStatisticsService.insert(userTradeStatistics);
}
}catch(Exception e){
logger.error("插入更新數(shù)據(jù)失敗", e);
return null;
}
}else{
logger.info("被鎖住了。。。。");
return null;
}
}
}
最后在業(yè)務(wù)層，調(diào)用線程執(zhí)行
TradeStatisticsTask task = new TradeStatisticsTask(userTradeStatisticsService,userTradeStatisticsRequestDTO);
O2oThreadPoolExecutor.getInstance().submit(task);
2、guava
前面三步和java一樣，只是在業(yè)務(wù)層處理層改成了guava的方式去處理線程的調(diào)用，使用guava調(diào)用是可以抓到線程運(yùn)行時(shí)的異常、并知道線程是否運(yùn)行成功；
TradeStatisticsTask task = new TradeStatisticsTask(userTradeStatisticsService,userTradeStatisticsRequestDTO);
// O2oThreadPoolExecutor.getInstance().submit(task);
//方法1：可以直接得到Future的返回值,或者處理錯(cuò)誤情況
ListeningExecutorService executorService = MoreExecutors.listeningDecorator(O2oThreadPoolExecutor.getInstance());
final ListenableFuture future = executorService.submit(task);
Futures.addCallback(future, new FutureCallback() {
@Override
public void onFailure(Throwable e) {
logger.error("線程執(zhí)行失敗", e);
}
@Override
public void onSuccess(Long message) {
logger.info("receive message :" + message);
}
});
//方法2：監(jiān)聽Future,對(duì)原生Future擴(kuò)展增強(qiáng),ListenFuture可以幫忙檢測Future是否執(zhí)行完成,如果完成就自動(dòng)調(diào)用回調(diào)函數(shù)
future.addListener(new Runnable(){
@Override
public void run() {
try {
logger.info("receive message :" + future.get());
} catch (Exception e) {//主要是InterruptedException、ExecutionException出錯(cuò)
logger.error("get message error :", e);
}
}
}, executorService);
