content: package com.alibaba.pokemon.bulbasaur.schedule.job;
  
  import java.util.concurrent.*;
  
  import com.alibaba.pokemon.bulbasaur.persist.domain.JobDO;
  import com.alibaba.pokemon.bulbasaur.schedule.ScheduleMachine;
  import com.alibaba.ttl.threadpool.TtlExecutors;
  import com.google.common.util.concurrent.ThreadFactoryBuilder;
  import com.alibaba.pokemon.bulbasaur.schedule.ScheduleMachineFactory;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * User:  yunche.ch ... (ว ˙o˙)ง
   * <p>
   * Date: 14-12-17
   * <p>
   * Time: 下午11:48
   */
  public abstract class AnsyJob implements Job {
      private static Logger logger = LoggerFactory.getLogger(AnsyJob.class);
      protected static ExecutorService executorService;
      protected JobDO jobDO;
      protected ScheduleMachineFactory scheduleMachineFactory;
      private static final int corePoolSize = 64;
      private static final int maximumPoolSize = 128;
  
      /**
       * 初始化Executor
       *
       * @author: yunche.ch@taobao.com
       * @since 2012-12-27 下午06:38:19
       */
      public static void initExecutor() {
          ThreadFactory namedThreadFactory = new ThreadFactoryBuilder().setNameFormat("AnsyJob-pool-%d").build();
  
          RejectedExecutionHandler rejectedExecutionHandler = new RejectedExecutionHandler() {
              @Override
              public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
                  if (!executor.isShutdown()) {
                      try {
                          executor.getQueue().put(r);
                      } catch (InterruptedException e) {
                          logger.error(String.format("线程池执行任务阻塞后异常,不可恢复! runnable=%s", r));
                      }
                  }
              }
          };
  
          ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(corePoolSize, maximumPoolSize,
              1L, TimeUnit.SECONDS, new LinkedBlockingQueue(), namedThreadFactory, rejectedExecutionHandler);
  
          /**
           * 在allowCoreThreadTimeOut设置为true时，ThreadPoolExecutor的keepAliveTime参数必须大于0。
           */
          threadPoolExecutor.allowCoreThreadTimeOut(true);
  
          /**
           * 增强 InheritableThreadLocal
           */
          executorService = TtlExecutors.getTtlExecutorService(threadPoolExecutor);
  
      }
  
      @Override
      public void doJob() {
  
          executorService.submit(new Runnable() {
              @Override
              public void run() {
                  logger.warn("========= AnsyJob ============ 开始执行Job！BizId = [" + jobDO.getBizId() + "]");
                  ScheduleMachine scheduleMachine = scheduleMachineFactory.newInstance(jobDO.getBizId());
                  scheduleMachine.run(jobDO.getStateName(), jobDO.getOutGoing());
                  logger.warn("======== AnsyJob ========== 执行job成功！");
              }
          });
      }
  
  }
  