java多线程任务生成

protected final void exeTaskList(final List<Task> tasklist) {
        if (tasklist == null || tasklist.size() == 0) {
            log.info(getTaskName() + ":当前没有需要同步 整帐户数据的任务!");
            return;
        }
        final SemaphoreCountDownExecutor executor = new SemaphoreCountDownExecutor(SEMAPHORE_PERMIT, tasklist.size());
        try {
            for (final Task task : tasklist) {
                try {
                    if (taskIsExecuting(task)) {//任务正在执行中
                        log.info("task:正在执行中，" + task);
                        executor.countDown();
                        continue;
                    }
                    synchronized (log) {
                        if (taskRuleService.canExec(task, taskService.findExecuting())) {
                            log.info("doTask after canExec" + task);
                            updateTaskExecuting(task);
                            log.info("doTask after updateExecuting" + task);
                            executor.submitTask(new Runnable() {
//                                @Transactional
                                @Override
                                public void run() {
                                    try {
                                        execute(task);//执行任务
                                        updateTaskSuccess(task);//更新任务状态
                                        log.info("doTask after updateSuccess" + task);
                                    } catch (Exception e) {
                                        log.error("executing error!", e);
                                        if (e.getMessage() != null) {
                                            taskService.updateErrorInfo(task.getId(), e.getMessage());
                                        }
                                        Integer count = task.getRetryCount();
                                        if (null != count && count.intValue() >= 5) {
                                            updateTaskFailed(task);
                                        } else {
                                            updateTaskRetry(task);
                                        }
                                    }
                                }
                            });
                        }
                    }
                } catch (Exception e) {//某个线程异常
                    log.error("exeTaskList exception"+e.getMessage(), e);
                }
            }
            log.info("exeTaskList before executor.await");
            executor.await(60 * 60);
        } catch (Exception e) {
            log.error(e.getMessage());
        } finally {
            log.info(getTaskName() + ", end");
        }
    }

public class SemaphoreCountDownExecutor {
	/**
	 * Logger for this class
	 */
	private static final Logger logger = LoggerFactory.getLogger(SemaphoreCountDownExecutor.class);

	private static final ExecutorService executor = 
		new ThreadPoolExecutor(8, 10000,60L, TimeUnit.SECONDS,new SynchronousQueue<Runnable>());
	private final Semaphore semaphore;
	private final CountDownLatch countDown;

	/**
	 * @param semaphorePermits :new Semaphore(semaphorePermits);
	 * @param count : new CountDownLatch(count);
	 */
	public SemaphoreCountDownExecutor(final int semaphorePermits,final int count) {
		super();
		this.semaphore = new Semaphore(semaphorePermits);
		this.countDown = new CountDownLatch(count);
	}

	/**
	 * <p>
	 * 堵塞，等待计数器为0 ，然后 关闭线程池
	 * </p>
	 * @author chenlong
	 * @version 2013-8-8
	 */
	public void await(final long timeSeconds) {
		try {
			if (timeSeconds <= 0){
				countDown.await(60 * 60, TimeUnit.SECONDS);
			}else{
				countDown.await(timeSeconds, TimeUnit.SECONDS);
			}
		} catch (InterruptedException e) {
			logger.error(e.getMessage());
		} 
		
	}

	/**
	 * <p>计数器减一</p>
	 */
	public void countDown(){
		countDown.countDown();
	}
	/***
	 * <p>
	 * 提交一个任务到带有信号变量的线程池，执行完后及异常情况释放信号变量且计数器自动减一
	 * </p>
	 * @param command
	 * @throws InterruptedException
	 */
	public void submitTask(final Runnable command) throws InterruptedException {
		semaphore.acquire(); // 接收一个信号量，此处可能抛出InterruptedException
		try {
			executor.execute(new Runnable() {
				@Override
				public void run() {
					// run中是同步执行
					try {
						command.run();
					} finally {
						semaphore.release();// 线程异常时释放信号量
						countDown.countDown();// 执行结束计数器-1
					}
				}
			});
		} catch (Exception e) {// 捕获线程池异常时， 释放信号及计数器-1
			logger.error(e.getMessage());
			countDown.countDown();// 执行结束计数器-1
			semaphore.release();
		}

	}

}
说明:
这里主要是从数据库里查询出当前需要执行的任务，然后判断当前任务是否能被执行，如果能被执行,则submit执行。
问题分析:
待续