java后台异步任务执行器TaskManager

 1 @Resource
 2 private TaskManager taskManager;
 3 
 4 public string  commit(TradeStatus status) {
 5     if (status== TradeStatus.UNDERWAY) {
 6 
 7     // 执行任务
 8     taskManager.addTask(new Runnable() {
 9 
10         @Override
11         public void run() {
12             handleUnderway(status); //运行业程序
13         }
14     });
15 
16     } else {
17                 
18     }
19 
20 }

 /**
     * 返回业务处理状态
     * 
     * 
     * <p>
     * 此方法需要被异步调用
     * </p>
     * 
     */
    public void handleUnderway(TradeSatus satus) {
        int waitTimeOut = 20*1000；
        try {
            if (logger.isDebugEnabled()) {

                logger.debug("提交查询支付状态");
            }

            Assert.notNull(trade, "交易对象不能为空");

            int timeOut = 0;

            String result = null;
            while (true) {

                result = queryStatus(satus); //获取状态

                if (result == TradeStatus.UNDERWAY) {
                    try {

                        if (timeOut > waitTimeOut) {

                            logger.warn(String.format("查询状态结果超时."));
                            break;
                        }

                        timeOut += 5000;
                        Thread.sleep(5000); // 5秒同步一次
                    } catch (InterruptedException e) {

                        if (logger.isInfoEnabled()) {
                            logger.info("等待支付结果被中断");
                        }
                        break;
                    }
                } else {
                    break;// 查到银行结果
                }
            }

            if (result != TradeStatus.UNDERWAY) {

                //处理结果
            }

            if (logger.isDebugEnabled()) {
                logger.debug(String.format("查询完成，查询结果为：%s", result));
            }

        } catch (Exception e) {
            
            logger.error(String.format("查询支付结果异常"), e);
            
            throw new SimpleException("查询支付结果异常", e);
        }

    }        

public interface TaskManager {

    /**
     * @param task
     */
    void addTask(Runnable task);

    /**
     * @return
     */
    int getActiveCount();
    
    /**
     * 停止任务管理器
     */
    void stop();

}

package com.zhonglian.jinjufin.service.impl;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import javax.annotation.Resource;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.core.task.TaskExecutor;

import com.service.TaskManager;

/**
 * 后台异步任务执行器
 *
 */
public class TaskManagerImpl implements TaskManager, InitializingBean {

    private static Log logger = LogFactory.getLog(TaskManagerImpl.class);

    private ThreadPoolExecutor executer = null;

    private int corePoolSize = 5;

    private int maximumPoolSize = 50;

    private long keepAliveTime = 10;

    private TimeUnit unit = TimeUnit.MINUTES;// 秒

    private BlockingQueue<Runnable> workQueue;

    private Thread t;
    
    @Resource
    private TaskExecutor taskExecutor; 

    @Override
    public void addTask(Runnable task) {

        taskExecutor.execute(task);
        
//        executer.execute(task);

    }

    @Override
    public int getActiveCount() {

        return 0;

    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
     */
    @Override
    public void afterPropertiesSet() throws Exception {

        if (executer == null) {
            
            workQueue = new LinkedBlockingQueue<Runnable>();

            executer = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, new ThreadPoolExecutor.AbortPolicy());
            
            
            t = new Thread(new Runnable() {
                
                @Override
                public void run() {

                    while (true) {
                        
                        try {
                            Thread.sleep(120 * 1000);
                        
                            if (logger.isDebugEnabled()) {
                                
                                String msg = String.format("队列大小:%d, 激活任务数: %d, poolSize: %d, 任务数：%d, 已处理任务数：%d", 
                                         workQueue.size(), getActiveCount(), executer.getPoolSize(), executer.getTaskCount(), executer.getCompletedTaskCount());
                                
                                logger.debug(msg);
                                
                            }
                        
                        } catch (InterruptedException e) {
                            return;
                        }
                        
                        
                    }
                    
                }
            });
            t.setDaemon(true);
            t.setName("异步任务健康检查线程");
            t.start();
            
        }
        
    }

    @Override
    public void stop() {

        if (executer == null)
            return;

        try {
            logger.info("准备停止任务管理器");
            executer.shutdown();
            executer.awaitTermination(5, TimeUnit.SECONDS);
        } catch (InterruptedException e) {

            logger.info("任务被中止.");
        } finally {
            if (!executer.isTerminated()) {
                logger.info("取消未完成的任务.");
            }
            executer.shutdownNow();
            logger.info("任务管理器停止完成.");
        }
    }

    /**
     * @return the corePoolSize
     */
    public int getCorePoolSize() {
        return corePoolSize;
    }

    /**
     * @param corePoolSize
     *            the corePoolSize to set
     */
    public void setCorePoolSize(int corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    /**
     * @return the maximumPoolSize
     */
    public int getMaximumPoolSize() {
        return maximumPoolSize;
    }

    /**
     * @param maximumPoolSize
     *            the maximumPoolSize to set
     */
    public void setMaximumPoolSize(int maximumPoolSize) {
        this.maximumPoolSize = maximumPoolSize;
    }

    /**
     * @return the keepAliveTime
     */
    public long getKeepAliveTime() {
        return keepAliveTime;
    }

    /**
     * @param keepAliveTime
     *            the keepAliveTime to set
     */
    public void setKeepAliveTime(long keepAliveTime) {
        this.keepAliveTime = keepAliveTime;
    }

    /**
     * @return the unit
     */
    public TimeUnit getUnit() {
        return unit;
    }

    /**
     * @param unit
     *            the unit to set
     */
    public void setUnit(TimeUnit unit) {
        this.unit = unit;
    }

}

       <!-- 异步任务执行器配置 -->
    <bean id="taskManager" class="com.service.impl.TaskManagerImpl">
        <property name="corePoolSize" value="30"></property>
        <property name="maximumPoolSize" value="100"></property>
        <property name="keepAliveTime" value="10"></property>
    </bean>
    
    <!-- 
        任务执行器  
        pool-size="5-20", 表示线程池活跃的线程数为5，最大线程数为20；
        queue-capacity="100"    表示队列大小为100
    -->
    <task:executor id="taskExecutor" keep-alive="30" pool-size="5-20" queue-capacity="100" rejection-policy="ABORT"/>