spring线程池(同步、异步）

<bean id="simpleAsyncTaskExecutor"
class="org.springframework.core.task.SimpleAsyncTaskExecutor">
<property name="daemon" value="true"/>
<property name="concurrencyLimit" value="2"/>
<property name="threadNamePrefix" value="simpleAsyncTaskExecutor"/>
</bean>

public class SimpleAsyncTaskExecutor extends CustomizableThreadCreator implements AsyncListenableTaskExecutor, Serializable {
//限流主要实现
private final SimpleAsyncTaskExecutor.ConcurrencyThrottleAdapter concurrencyThrottle = new SimpleAsyncTaskExecutor.ConcurrencyThrottleAdapter();
private ThreadFactory threadFactory;
//设置最大的线程数量
public void setConcurrencyLimit(int concurrencyLimit) {
this.concurrencyThrottle.setConcurrencyLimit(concurrencyLimit);
}
//是否开启了限流 限流数量大于0？
public final boolean isThrottleActive() {
return this.concurrencyThrottle.isThrottleActive();
}
//1.是否开启限流 否则不开启限流处理
//2.执行开始之前检测是否可以满足要求 当前数量++
//3.开启限流将执行的Runable进行封装，执行完成调用final方法 当前数量--
public void execute(Runnable task, long startTimeout) {
Assert.notNull(task, "Runnable must not be null");
if(this.isThrottleActive() && startTimeout > 0L) {
this.concurrencyThrottle.beforeAccess();
this.doExecute(new SimpleAsyncTaskExecutor.ConcurrencyThrottlingRunnable(task));
} else {
this.doExecute(task);
}
}
//异步提交有返回值
public Future<?> submit(Runnable task) {
FutureTask future = new FutureTask(task, (Object)null);
this.execute(future, 9223372036854775807L);
return future;
} public <T> Future<T> submit(Callable<T> task) {
FutureTask future = new FutureTask(task);
this.execute(future, 9223372036854775807L);
return future;
} public ListenableFuture<?> submitListenable(Runnable task) {
ListenableFutureTask future = new ListenableFutureTask(task, (Object)null);
this.execute(future, 9223372036854775807L);
return future;
} public <T> ListenableFuture<T> submitListenable(Callable<T> task) {
ListenableFutureTask future = new ListenableFutureTask(task);
this.execute(future, 9223372036854775807L);
return future;
}
//拥有工厂？没有的话调用父类可以设置各种参数的创建线程
protected void doExecute(Runnable task) {
Thread thread = this.threadFactory != null?this.threadFactory.newThread(task):this.createThread(task);
thread.start();
}
//父类的方法，方便配置线程，方便xml设置线程参数CustomizableThreadCreator
public Thread createThread(Runnable runnable) {
Thread thread = new Thread(getThreadGroup(), runnable, nextThreadName());
thread.setPriority(getThreadPriority());
thread.setDaemon(isDaemon());
return thread;
} }

//下面只是对于操作进行简单的封装，最真的实现还是抽象的ConcurrencyThrottleSupport
private static class ConcurrencyThrottleAdapter extends ConcurrencyThrottleSupport {
private ConcurrencyThrottleAdapter() {
} protected void beforeAccess() {
super.beforeAccess();
} protected void afterAccess() {
super.afterAccess();
}
}

//这里是对于Runable对象执行在次封装，在执行完毕后处理限流操作
private class ConcurrencyThrottlingRunnable implements Runnable {
private final Runnable target; public ConcurrencyThrottlingRunnable(Runnable target) {
this.target = target;
} public void run() {
try {
this.target.run();
} finally {
SimpleAsyncTaskExecutor.this.concurrencyThrottle.afterAccess();
} }
}

protected void doExecute(Runnable task) {
Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
thread.start();
}

public Thread createThread(Runnable runnable) {
Thread thread = new Thread(getThreadGroup(), runnable, nextThreadName());
thread.setPriority(getThreadPriority());
thread.setDaemon(isDaemon());
return thread;
}

public interface SuccessCallback<T> { /**
* Called when the {@link ListenableFuture} successfully completes.
* @param result the result
*/
void onSuccess(T result);
}
public interface FailureCallback { /**
* Called when the {@link ListenableFuture} fails to complete.
* @param ex the exception that triggered the failure
*/
void onFailure(Throwable ex);
}public interface ListenableFuture<T> extends Future<T> {
//成功和失败的集合
void addCallback(ListenableFutureCallback<? super T> var1); void addCallback(SuccessCallback<? super T> var1, FailureCallback var2);
}

public class ListenableFutureTask<T> extends FutureTask<T> implements ListenableFuture<T> {
private final ListenableFutureCallbackRegistry<T> callbacks = new ListenableFutureCallbackRegistry(); public ListenableFutureTask(Callable<T> callable) {
super(callable);
} public ListenableFutureTask(Runnable runnable, T result) {
super(runnable, result);
} public void addCallback(ListenableFutureCallback<? super T> callback) {
this.callbacks.addCallback(callback);
} public void addCallback(SuccessCallback<? super T> successCallback, FailureCallback failureCallback) {
this.callbacks.addSuccessCallback(successCallback);
this.callbacks.addFailureCallback(failureCallback);
}
//FutureTask执行完成后的回调，调用监听接口的实现类的方法
protected final void done() {
Object cause;
try {
Object ex = this.get();
//回调实现类的方法
this.callbacks.success(ex);
return;
} catch (InterruptedException var3) {
Thread.currentThread().interrupt();
return;
} catch (ExecutionException var4) {
cause = var4.getCause();
if(cause == null) {
cause = var4;
}
} catch (Throwable var5) {
cause = var5;
} this.callbacks.failure((Throwable)cause);
}
}

public class ListenableFutureCallbackRegistry<T> { private final Queue<SuccessCallback<? super T>> successCallbacks = new LinkedList<SuccessCallback<? super T>>(); private final Queue<FailureCallback> failureCallbacks = new LinkedList<FailureCallback>(); private State state = State.NEW; private Object result = null; private final Object mutex = new Object(); /**
* Add the given callback to this registry.
* @param callback the callback to add
*/
public void addCallback(ListenableFutureCallback<? super T> callback) {
Assert.notNull(callback, "'callback' must not be null");
synchronized (this.mutex) {
switch (this.state) {
case NEW:
this.successCallbacks.add(callback);
this.failureCallbacks.add(callback);
break;
case SUCCESS:
callback.onSuccess((T) this.result);
break;
case FAILURE:
callback.onFailure((Throwable) this.result);
break;
}
}
} /**
* Add the given success callback to this registry.
* @param callback the success callback to add
* @since 4.1
*/
public void addSuccessCallback(SuccessCallback<? super T> callback) {
Assert.notNull(callback, "'callback' must not be null");
synchronized (this.mutex) {
switch (this.state) {
case NEW:
this.successCallbacks.add(callback);
break;
case SUCCESS:
callback.onSuccess((T) this.result);
break;
}
}
} /**
* Add the given failure callback to this registry.
* @param callback the failure callback to add
* @since 4.1
*/
public void addFailureCallback(FailureCallback callback) {
Assert.notNull(callback, "'callback' must not be null");
synchronized (this.mutex) {
switch (this.state) {
case NEW:
this.failureCallbacks.add(callback);
break;
case FAILURE:
callback.onFailure((Throwable) this.result);
break;
}
}
} /**
* Trigger a {@link ListenableFutureCallback#onSuccess(Object)} call on all
* added callbacks with the given result.
* @param result the result to trigger the callbacks with
*/
public void success(T result) {
synchronized (this.mutex) {
this.state = State.SUCCESS;
this.result = result;
while (!this.successCallbacks.isEmpty()) {
this.successCallbacks.poll().onSuccess(result);
}
}
} public void failure(Throwable ex) {
synchronized (this.mutex) {
this.state = State.FAILURE;
this.result = ex;
while (!this.failureCallbacks.isEmpty()) {
this.failureCallbacks.poll().onFailure(ex);
}
}
} private enum State {NEW, SUCCESS, FAILURE}}

<bean id="taskExecutor" class="org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor">
<property name="corePoolSize" value="5" />
<property name="maxPoolSize" value="10" />
<property name="queueCapacity" value="25" />
</bean><bean id="taskExecutorExample" class="TaskExecutorExample">
<constructor-arg ref="taskExecutor" />
</bean>

import org.springframework.core.task.TaskExecutor;public class TaskExecutorExample { private class MessagePrinterTask implements Runnable { private String message; public MessagePrinterTask(String message) {
this.message = message;
} public void run() {
System.out.println(message);
} } private TaskExecutor taskExecutor; public TaskExecutorExample(TaskExecutor taskExecutor) {
this.taskExecutor = taskExecutor;
} public void printMessages() {
for(int i = 0; i < 25; i++) {
taskExecutor.execute(new MessagePrinterTask("Message" + i));
}
}
}

<?xml version="1.0" encoding="UTF-8"?>
<!--Spring框架的xml标签定义文档, 可访问http://www.springframework.org/schema/task/查看最新task组件的xml标签文档-->
<beans xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="
http://www.springframework.org/schema/task
http://www.springframework.org/schema/task/spring-task-3.2.xsd">
<!--扫描项目实例化@Component，@Service，@Controller修饰的类-->
<context:component-scan base-package="com.your_app" /> <!--create a SimpleAsyncTaskExecutor instance-->
<task:annotation-driven/>
</beans>

<?xml version="1.0" encoding="UTF-8"?>
<!--Spring框架的xml标签定义文档, 可访问http://www.springframework.org/schema/task/查看最新task组件的xml标签文档-->
<beans xmlns:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="
http://www.springframework.org/schema/task
http://www.springframework.org/schema/task/spring-task-3.2.xsd"> <!--扫描项目实例化@Component，@Service，@Controller修饰的类-->
<context:component-scan base-package="com.your_app" /> <!-- 在代码中@Async不加参数就会使用task:annotation-driven标签定义的executor-->
<task:annotation-driven executor="myExecutor"/>
<!-- 在代码中@Async("myExecutor")可以显式指定executor为"myExecutor"-->
<task:executor id="myExecutor"
pool-size="5-25"
queue-capacity="100"
rejection-policy="CALLER_RUNS"/>
</beans>