java线程池增强实现

       前两天在开涛的公众号里，开涛聊到一次请求生成唯一的traceId在各个业务系统中传递，然后通过日志收集各个业务服务中的日志，形成一次请求的完整日志。开涛简单的提到了是使用自己实现的线程池增强技术来传递traceId。

       我这边系统也有类似的需求。所以我就尝试性地实现了下线程池增强。本来想着既然是增强，第一反应是用代理技术去实现，后来发现不需要代理就可以简单地实现。

       我大致的场景是把线程中附带的用户信息从当前线程传递到要开启的新的线程中去，并告诉MDC进行日志打印。    

       下面简单介绍下代码的实现：

       首先是自定义的线程池UserContextThreadPoolExecutor.java类，继承了jdk并发包中的ThreadPoolExecutor类：

       

package com.onlyou.framework.ext.concurrent;

import com.onlyou.olyfinance.common.UserContext;
import java.util.concurrent.*;

public class UserContextThreadPoolExecutor extends ThreadPoolExecutor {

	public UserContextThreadPoolExecutor(int corePoolSize,
	                                int maximumPoolSize,
	                                long keepAliveTime,
	                                TimeUnit unit,
	                                BlockingQueue<Runnable> workQueue) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
	}

	public UserContextThreadPoolExecutor(int corePoolSize,
	                                int maximumPoolSize,
	                                long keepAliveTime,
	                                TimeUnit unit,
	                                BlockingQueue<Runnable> workQueue,
	                                ThreadFactory threadFactory) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,
				threadFactory);
	}

	public UserContextThreadPoolExecutor(int corePoolSize,
	                                int maximumPoolSize,
	                                long keepAliveTime,
	                                TimeUnit unit,
	                                BlockingQueue<Runnable> workQueue,
	                                RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
	}

	public UserContextThreadPoolExecutor(int corePoolSize,
	                                int maximumPoolSize,
	                                long keepAliveTime,
	                                TimeUnit unit,
	                                BlockingQueue<Runnable> workQueue,
	                                ThreadFactory threadFactory,
	                                RejectedExecutionHandler handler) {
		super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
	}

	/**
	 * 重写执行线程实例的方法
	 * @param command
	 */
	public void execute(Runnable command) {
		if (command == null){
			throw new NullPointerException();
		}
        UserContextRunnableTask task =new UserContextRunnableTask();
		task.setUser(UserContext.getUser());
		task.setDelegate(command);
		super.execute(task);
	}
}

   UserContext.getUser()是从线程中获取到用户信息，实际上扔给线程池的类是UserContextRunnableTask类：

package com.onlyou.framework.ext.concurrent;

import com.onlyou.olyfinance.application.vo.UserInfo;
import com.onlyou.olyfinance.common.UserContext;

public class UserContextRunnableTask implements Runnable{

	private UserInfo user;

	private Runnable delegate;

	@Override
	public void run() {
		UserContext.setUser(user);
		UserContext.setMDCBeforeInvoke();
		try{
			delegate.run();
		}catch (Throwable e){
			throw e;
		}finally {
			UserContext.removeUser();
			UserContext.removeMDCAfterInvoke();
		}
	}

	public void setUser(UserInfo user) {
		this.user = user;
	}

	public void setDelegate(Runnable delegate) {
		this.delegate = delegate;
	}
}

    多线程中要执行的业务内容是UserContextRunnableTask 中的Runnable delegate去执行的delegate.run()。而真正线程池执行的是UserContextRunnableTask.run()。 在UserContextRunnableTask的run()方法中我们UserContext.setUser(user)里我们把用户信息绑定到当前线程中（基于线程变量），并在线程执行完UserContext.removeUser()把用户信息从当前线程中移除。

    这里可能有人会有疑问，我只是重写了执行Runnable接口的多线程实例。那JDK1.5后出来的callable接口呢。其实callable接口的实例扔给线程池，线程池会调用这个方法：

protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        return new FutureTask<T>(callable);
    }

 返回一个FutureTask实例给线程池。这个FutureTask其实就是实现了Runnable接口。在FutureTask中的run()方法中：

public void run() {
        if (state != NEW ||
            !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                         null, Thread.currentThread()))
            return;
        try {
            Callable<V> c = callable;
            if (c != null && state == NEW) {
                V result;
                boolean ran;
                try {
                    result = c.call();
                    ran = true;
                } catch (Throwable ex) {
                    result = null;
                    ran = false;
                    setException(ex);
                }
                if (ran)
                    set(result);
            }
        } finally {
            // runner must be non-null until state is settled to
            // prevent concurrent calls to run()
            runner = null;
            // state must be re-read after nulling runner to prevent
            // leaked interrupts
            int s = state;
            if (s >= INTERRUPTING)
                handlePossibleCancellationInterrupt(s);
        }
    }

       在FutureTask的run()方法中，会去调用callable.call()并做一些额外的处理， result = c.call()，result会赋值给FutureTask的outcome属性。outcome就是返回值，用get()即可以获取。