1、Java

當要使用線程去執行一個任務時，可以使用ExecutorService.submit(new Callable);

這樣可以不影響其他的業務的執行，但是在線程中的異常不能捕獲到，也不能知道線程的執行情況；

首先創建一個固定大小堵塞的線程池：

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);

}

}

然后實現RejectedExecutionHandler，主要是當拒絕之后，重新put進去，防止丟失

public class RejectedExecutionHandlerImpl implements RejectedExecutionHandler {

@Override

public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {

if(!executor.isShutdown()){

try{

executor.getQueue().put(r);

}catch(InterruptedException e){

}

}

}

}

在然后定義線程需要執行的任務，實現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任務："+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{

//先判斷在統計數據中是否有該用戶、銀行卡、當天的記錄，如果沒有，則插入，否則更新

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("開始插入統計數據");

UserTradeStatistics userTradeStatistics = new UserTradeStatistics();

BeanMapper.copy(this.userTradeStatisticsRequestDTO, userTradeStatistics);

return userTradeStatisticsService.insert(userTradeStatistics);

}

}catch(Exception e){

logger.error("插入更新數據失敗", e);

return null;

}

}else{

logger.info("被鎖住了。。。。");

return null;

}

}

}

最后在業務層，調用線程執行

TradeStatisticsTask task = new TradeStatisticsTask(userTradeStatisticsService,userTradeStatisticsRequestDTO);

O2oThreadPoolExecutor.getInstance().submit(task);

2、guava

前面三步和java一樣，只是在業務層處理層改成了guava的方式去處理線程的調用，使用guava調用是可以抓到線程運行時的異常、并知道線程是否運行成功；

TradeStatisticsTask task = new TradeStatisticsTask(userTradeStatisticsService,userTradeStatisticsRequestDTO);

// O2oThreadPoolExecutor.getInstance().submit(task);

//方法1：可以直接得到Future的返回值,或者處理錯誤情況

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("線程執行失敗", e);

}

@Override

public void onSuccess(Long message) {

logger.info("receive message :" + message);

}

});

//方法2：監聽Future,對原生Future擴展增強,ListenFuture可以幫忙檢測Future是否執行完成,如果完成就自動調用回調函數

future.addListener(new Runnable(){

@Override

public void run() {

try {

logger.info("receive message :" + future.get());

} catch (Exception e) {//主要是InterruptedException、ExecutionException出錯

logger.error("get message error :", e);

}

}

}, executorService);

總結

以上是生活随笔為你收集整理的线程Java的两种方式_多线程(java和guava两种方式)：的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。