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