process task by queue

public abstract class TaskQueue<T> {

	ExecutorService executor = null;
	
	ArrayBlockingQueue<T> queue;
	
	private volatile boolean running = false;	
	
	protected abstract void doWork(T item);

	public  TaskQueue(ExecutorService executor, ArrayBlockingQueue<T> queue){
		this.executor = executor;
		this.queue = queue;
		start();
	}
	

	// start on Server startup
	public void start() {
		running = true;		
		executor.submit(new Runnable() {
			@Override
			public void run() {
				processTask();
			}
		});

		Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
			@Override
			public void run() {
				stop();
			}
		}));
	}
	
	private void processTask() {
		try {
			T item = null;
			while(running){
				item = queue.poll(5, TimeUnit.SECONDS);
				if(item != null){
					doWork(item);
				}
			}
			
			// process left items on queue
			while((item = queue.poll() ) != null){
				doWork(item);
			}
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
	}

	public void stop() {
		try {
			running = false;
//			long start = System.currentTimeMillis();
			executor.shutdown();
			executor.awaitTermination(30, TimeUnit.SECONDS);
//			long cost = System.currentTimeMillis() - start;
//			System.out.println("cost time " + cost + "ms");
			executor = null;
		} catch (InterruptedException e) {
			Thread.currentThread().interrupt();
		}
	}

	public void put(T item) {
		try {
//			System.out.println("put answer " + survey.getTitle());
			queue.put(item);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}

}

实现类

public class AnswerStatQueue extends TaskQueue<Answer>{

	AnswerService answerService = new AnswerService();
	
	private AnswerStatQueue(ExecutorService executor, ArrayBlockingQueue<Answer> queue) {
		super(executor, queue);
	}

	@Override
	protected void doWork(Answer answer) {
//		System.out.println("procss answer " + answer.getTitle());
		answerService.incrSurveyStat(answer);
//		System.out.println("finish answer ......!!!");
	}
	
	
	//***************************single instance *****************************
	
	
	private static volatile AnswerStatQueue instance;

	public static AnswerStatQueue getInstance() {
		if(instance == null){
			synchronized (AnswerStatQueue.class) {
				if(instance == null){
					instance = createAnswerStateQueue();
				}
			}
		}
		return instance;
	}
	
	private static AnswerStatQueue createAnswerStateQueue(){
		ExecutorService executor = Executors.newSingleThreadExecutor();
		ArrayBlockingQueue<Answer> queue = new ArrayBlockingQueue<Answer>(10000);
		instance = new AnswerStatQueue(executor, queue);
		return instance;
	}

}