public void execute(Runnable task, long startTimeout) {
  Assert.notNull(task, "Runnable must not be null");
  Runnable taskToUse = (this.taskDecorator != null ? this.taskDecorator.decorate(task) : task);
  //判斷是否開啟限流機(jī)制
  if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
   //執(zhí)行前置操作，進(jìn)行限流
   this.concurrencyThrottle.beforeAccess();
   //執(zhí)行完線程任務(wù)，會執(zhí)行后置操作concurrencyThrottle.afterAccess()，配合進(jìn)行限流
   doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
  }
  else {
   doExecute(taskToUse);
  }
}

protected void beforeAccess() {
	if (this.concurrencyLimit == NO_CONCURRENCY) {
		throw new IllegalStateException(
				"Currently no invocations allowed - concurrency limit set to NO_CONCURRENCY");
	}
	if (this.concurrencyLimit > 0) {
		boolean debug = logger.isDebugEnabled();
		synchronized (this.monitor) {
			boolean interrupted = false;
			while (this.concurrencyCount >= this.concurrencyLimit) {
				if (interrupted) {
					throw new IllegalStateException("Thread was interrupted while waiting for invocation access, " +
							"but concurrency limit still does not allow for entering");
				}
				if (debug) {
					logger.debug("Concurrency count " + this.concurrencyCount +
							" has reached limit " + this.concurrencyLimit + " - blocking");
				}
				try {
					this.monitor.wait();
				}
				catch (InterruptedException ex) {
					// Re-interrupt current thread, to allow other threads to react.
					Thread.currentThread().interrupt();
					interrupted = true;
				}
			}
			if (debug) {
				logger.debug("Entering throttle at concurrency count " + this.concurrencyCount);
			}
			this.concurrencyCount++;
		}
	}
}

protected void beforeAccess() {
	if (this.concurrencyLimit == NO_CONCURRENCY) {
		throw new IllegalStateException(
				"Currently no invocations allowed - concurrency limit set to NO_CONCURRENCY");
	}
	if (this.concurrencyLimit > 0) {
		boolean debug = logger.isDebugEnabled();
		synchronized (this.monitor) {
			boolean interrupted = false;
			while (this.concurrencyCount >= this.concurrencyLimit) {
				if (interrupted) {
					throw new IllegalStateException("Thread was interrupted while waiting for invocation access, " +
							"but concurrency limit still does not allow for entering");
				}
				if (debug) {
					logger.debug("Concurrency count " + this.concurrencyCount +
							" has reached limit " + this.concurrencyLimit + " - blocking");
				}
				try {
					this.monitor.wait();
				}
				catch (InterruptedException ex) {
					// Re-interrupt current thread, to allow other threads to react.
					Thread.currentThread().interrupt();
					interrupted = true;
				}
			}
			if (debug) {
				logger.debug("Entering throttle at concurrency count " + this.concurrencyCount);
			}
			this.concurrencyCount++;
		}
	}
}

protected void afterAccess() {
	if (this.concurrencyLimit >= 0) {
		synchronized (this.monitor) {
			this.concurrencyCount--;
			if (logger.isDebugEnabled()) {
				logger.debug("Returning from throttle at concurrency count " + this.concurrencyCount);
			}
			this.monitor.notify();
		}
	}
}

@Configuration
@EnableAsync
public class AsyncCommonConfig extends AsyncConfigurerSupport {
  @Override
  public Executor getAsyncExecutor() {
    SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor();
    //設(shè)置允許同時執(zhí)行的線程數(shù)為10
 executor.setConcurrencyLimit(10);
    return executor;
  }
}

public class AsyncConfig implements AsyncConfigurer{
  private Logger logger = LogManager.getLogger();

  @Value("${thread.pool.corePoolSize:10}")
  private int corePoolSize;

  @Value("${thread.pool.maxPoolSize:20}")
  private int maxPoolSize;

  @Value("${thread.pool.keepAliveSeconds:4}")
  private int keepAliveSeconds;

  @Value("${thread.pool.queueCapacity:512}")
  private int queueCapacity;

  @Override
  public Executor getAsyncExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(corePoolSize);
    executor.setMaxPoolSize(maxPoolSize);
    executor.setKeepAliveSeconds(keepAliveSeconds);
    executor.setQueueCapacity(queueCapacity);
    executor.setRejectedExecutionHandler((Runnable r, ThreadPoolExecutor exe) -> {
        logger.warn("當(dāng)前任務(wù)線程池隊(duì)列已滿.");
    });
    executor.initialize();
    return executor;
  }

  @Override
  public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
    return new AsyncUncaughtExceptionHandler() {
      @Override
      public void handleUncaughtException(Throwable ex , Method method , Object... params) {
        logger.error("線程池執(zhí)行任務(wù)發(fā)生未知異常.", ex);
      }
    };
  }
}