Java多线程编程实战指南（设计模式篇，黄文海）-之管道线模式

标签：

不得不说，本人工作上很少有使用多线程编程技术的地方。由于本人工作上经常使用的是类似SSH等框架搭建MVC架构，所以更加习惯于编写一些优秀程序员所唾弃的样板式的代码。最近看了文海的多线程编程实战指南，瞬间眼前一亮。觉得有很多自己可以学习的，事实上，我已经在最近的项目中使用上了那本书介绍的两相终止模式、串行封闭模式、生产者消费者模式以及线程池等技术，确实在许多方面改进了我们服务端的吞吐量。说到这里本人吐槽一下，由于是毕业后转行，目前也才工作一年还不满2个月。所以原谅我的得瑟，但我相信我以后会做的更好！

个人觉的那是一本实战性极强的书，很有阅读价值。但我本人不是来打广告的。废话少说，进入正题。我今天要讲的是我对书中的管道线模式的理解。所谓管道线模式，我的理解就是将一个任务分成多个阶段，任务的完成需要经过所有的阶段才能完成。然后，倘若采用多线程的模式，使得任务的不同阶段在不同的线程中完成。那么，客户端或客户端代码提交的任务只要在阶段一处理完成就返回了，尽管该任务或许还在处理中。这样的好处呢就是减少了客户端代码的等待。但事实上任何一个具体的任务依然是串行执行的，这样可以避免某些令人纠结的线程安全问题。但假如客户端代码同时并发的提交了多个任务，那么处理任务一的阶段一的线程在处理完这个具体阶段后就可以紧接着处理任务二的阶段一，尽管任务一可能还处于阶段二或者阶段三等。所以，在多任务提交的情况下。管道线模式又有并发执行的效果，我们姑且称为伪并发吧。

之所以写下这篇博客，是因为管道线模式涉及的其他模式太多，这个模式下可以涉及线程池的使用，能找到责任链模式的影子，可以运用上装饰器 模式，以及多线程编程中的两相终止模式等等。所以有利于巩固自己所学，并增加自己的综合实战能力。哈哈，不说了，我们来具体的看代码吧。

首先，这是对各个管道的抽象，由于任务是在各个管道中顺序处理的，任务在一个管道中处理到某个阶段后必然要流转到下一个管道，所以接口中定义了设置下一个管道的setNextPipe()方法等，还在初始化方法中定义了如何传递上下文信息等，其余不再赘述：

<span style="font-size:18px;">public interface Pipe<IN, OUT>
{
	/**
	 * 设置当前Pipe实例的下一个Pipe实例。
	 * @param nextPipe
	 *            下一个Pipe实例
	 */
	void setNextPipe(Pipe<?, ?> nextPipe);

	/**
	 * 初始化当前Pipe实例对外提供的服务。
	 * @param pipeCtx
	 */
	void init(PipeContext pipeCtx);

	/**
	 * 停止当前Pipe实例对外提供的服务。
	 * 
	 * @param timeout
	 * @param unit
	 */
	void shutdown(long timeout, TimeUnit unit);

	/**
	 * 对输入元素进行处理，并将处理结果作为下一个Pipe实例的输入。
	 */
	void process(IN input) throws InterruptedException;
}
</span>

<span style="font-size:18px;">public interface PipeContext
{
	/**
	 * 用于对处理阶段抛出的异常进行处理.
	 * 
	 * @param exp
	 */
	void handleError(PipeException exp);
}
</span>

<span style="font-size:18px;">public class PipeException extends Exception
{
	private static final long serialVersionUID = -2944728968269016114L;
	/**
	 * 抛出异常的Pipe实例。
	 */
	public final Pipe<?, ?> sourcePipe;

	/**
	 * 抛出异常的Pipe实例在抛出异常时所处理的输入元素。
	 */
	public final Object input;

	public PipeException(Pipe<?, ?> sourcePipe, Object input, String message)
	{
		super(message);
		this.sourcePipe = sourcePipe;
		this.input = input;
	}

	public PipeException(Pipe<?, ?> sourcePipe, Object input, String message, Throwable cause)
	{
		super(message, cause);
		this.sourcePipe = sourcePipe;
		this.input = input;
	}
}</span>

<span style="font-size:18px;">public abstract class AbstractPipe<IN, OUT> implements Pipe<IN, OUT>
{
	protected volatile Pipe<?, ?> nextPipe = null;
	protected volatile PipeContext pipeCtx;

	@Override
	public void init(PipeContext pipeCtx)
	{
		this.pipeCtx = pipeCtx;

	}

	@Override
	public void setNextPipe(Pipe<?, ?> nextPipe)
	{
		this.nextPipe = nextPipe;

	}

	@Override
	public void shutdown(long timeout, TimeUnit unit)
	{
		// 什么也不做
	}

	/**
	 * 留给子类实现。用于子类实现其任务处理逻辑。
	 * 
	 * @param input
	 *            输入元素（任务）
	 * @return 任务的处理结果
	 * @throws PipeException
	 */
	protected abstract OUT doProcess(IN input) throws PipeException;

	@SuppressWarnings("unchecked")
	public void process(IN input) throws InterruptedException
	{

		try
		{

			OUT out = doProcess(input);
			if (null != nextPipe)
			{
				if (null != out)
				{
					((Pipe<OUT, ?>) nextPipe).process(out);
				}
			}
		}
		catch (InterruptedException e)
		{
			Thread.currentThread().interrupt();
		}
		catch (PipeException e)
		{
			pipeCtx.handleError(e);
		}
	}
}
</span>

既然任务是要在各个管道中顺序处理的，一个管道根据输入得到了某种中间结果，并作为下一个管道的输入，直到得到最终结果，那么，这些管道是怎么串联起来的呢？接下来的接口就是定义如何串联各个管道的管道线接口，仔细看一看并好好想想吧，是不是能找到责任链模式的影子呢，哈哈。

<span style="font-size:18px;">public interface Pipeline<IN, OUT> extends Pipe<IN, OUT>
{

	/**
	 * 往该Pipeline实例中添加一个Pipe实例。
	 * 
	 * @param pipe
	 *            Pipe实例
	 */
	void addPipe(Pipe<?, ?> pipe);
}
</span>

<span style="font-size:18px;">public class SimplePipeline<T, OUT> extends AbstractPipe<T, OUT> implements Pipeline<T, OUT>
{

	private final Queue<Pipe<?, ?>> pipes = new LinkedList<Pipe<?, ?>>();

	private final ExecutorService helperExecutor;

	public SimplePipeline()
	{
		this(Executors.newSingleThreadExecutor(new ThreadFactory()
		{

			@Override
			public Thread newThread(Runnable r)
			{
				Thread t = new Thread(r, "SimplePipeline-Helper");
				t.setDaemon(true);
				return t;
			}

		}));

	}

	public SimplePipeline(final ExecutorService helperExecutor)
	{
		super();
		this.helperExecutor = helperExecutor;
	}

	@Override
	public void shutdown(long timeout, TimeUnit unit)
	{
		Pipe<?, ?> pipe;

		while (null != (pipe = pipes.poll()))
		{
			pipe.shutdown(timeout, unit);
		}

		helperExecutor.shutdown();

	}

	@Override
	protected OUT doProcess(T input) throws PipeException
	{
		// 什么也不做
		return null;
	}

	@Override
	public void addPipe(Pipe<?, ?> pipe)
	{

		// Pipe间的关联关系在init方法中建立
		pipes.add(pipe);
	}

	public <INPUT, OUTPUT> void addAsWorkerThreadBasedPipe(Pipe<INPUT, OUTPUT> delegate, int workerCount)
	{
		addPipe(new WorkerThreadPipeDecorator<INPUT, OUTPUT>(delegate, workerCount));
	}

	public <INPUT, OUTPUT> void addAsThreadPoolBasedPipe(Pipe<INPUT, OUTPUT> delegate, ExecutorService executorSerivce)
	{
		addPipe(new ThreadPoolPipeDecorator<INPUT, OUTPUT>(delegate, executorSerivce));
	}

	@Override
	public void process(T input) throws InterruptedException
	{

		@SuppressWarnings("unchecked")
		Pipe<T, ?> firstPipe = (Pipe<T, ?>) pipes.peek();

		firstPipe.process(input);
	}

	@Override
	public void init(final PipeContext ctx)
	{

		LinkedList<Pipe<?, ?>> pipesList = (LinkedList<Pipe<?, ?>>) pipes;
		Pipe<?, ?> prevPipe = this;
		for (Pipe<?, ?> pipe : pipesList)
		{
			prevPipe.setNextPipe(pipe);
			prevPipe = pipe;
		}

		Runnable task = new Runnable()
		{
			@Override
			public void run()
			{
				for (Pipe<?, ?> pipe : pipes)
				{
					pipe.init(ctx);
				}

			}

		};

		helperExecutor.submit(task);
	}

	public PipeContext newDefaultPipelineContext()
	{
		return new PipeContext()
		{
			@Override
			public void handleError(final PipeException exp)
			{
				helperExecutor.submit(new Runnable()
				{

					@Override
					public void run()
					{
						exp.printStackTrace();
					}

				});
			}

		};
	}
}
</span>

<span style="font-size:18px;">public class ThreadPoolPipeDecorator<IN, OUT> implements Pipe<IN, OUT>
{
	private final Pipe<IN, OUT> delegate;
	private final ExecutorService executorSerivce;

	// 线程池停止标志。
	private final TerminationToken terminationToken;
	private final CountDownLatch stageProcessDoneLatch = new CountDownLatch(1);

	public ThreadPoolPipeDecorator(Pipe<IN, OUT> delegate, ExecutorService executorSerivce)
	{
		this.delegate = delegate;
		this.executorSerivce = executorSerivce;
		this.terminationToken = TerminationToken.newInstance(executorSerivce);
	}

	@Override
	public void init(PipeContext pipeCtx)
	{
		delegate.init(pipeCtx);

	}

	@Override
	public void process(final IN input) throws InterruptedException
	{

		Runnable task = new Runnable()
		{
			@Override
			public void run()
			{
				int remainingReservations = -1;
				try
				{
					delegate.process(input);
				}
				catch (InterruptedException e)
				{
					;
				}
				finally
				{
					remainingReservations = terminationToken.reservations.decrementAndGet();
				}

				if (terminationToken.isToShutdown() && 0 == remainingReservations)
				{
					stageProcessDoneLatch.countDown();
				}
			}

		};

		executorSerivce.submit(task);

		terminationToken.reservations.incrementAndGet();

	}

	@Override
	public void shutdown(long timeout, TimeUnit unit)
	{
		terminationToken.setIsToShutdown();

		if (terminationToken.reservations.get() > 0)
		{
			try
			{
				if (stageProcessDoneLatch.getCount() > 0)
				{
					stageProcessDoneLatch.await(timeout, unit);
				}
			}
			catch (InterruptedException e)
			{
				;
			}
		}

		delegate.shutdown(timeout, unit);
	}

	@Override
	public void setNextPipe(Pipe<?, ?> nextPipe)
	{
		delegate.setNextPipe(nextPipe);
	}

	/**
	 * 线程池停止标志。 每个ExecutorService实例对应唯一的一个TerminationToken实例。 这里使用了Two-phase
	 * Termination模式（第5章）的思想来停止多个Pipe实例所共用的 线程池实例。
	 * 
	 * @author Viscent Huang
	 *
	 */
	private static class TerminationToken extends io.github.viscent.mtpattern.ch5.tpt.TerminationToken
	{
		private final static ConcurrentMap<ExecutorService, TerminationToken> INSTANCES_MAP = new ConcurrentHashMap<ExecutorService, TerminationToken>();

		// 私有构造器
		private TerminationToken()
		{

		}

		void setIsToShutdown()
		{
			this.toShutdown = true;
		}

		static TerminationToken newInstance(ExecutorService executorSerivce)
		{
			TerminationToken token = INSTANCES_MAP.get(executorSerivce);
			if (null == token)
			{
				token = new TerminationToken();
				TerminationToken existingToken = INSTANCES_MAP.putIfAbsent(executorSerivce, token);
				if (null != existingToken)
				{
					token = existingToken;
				}
			}
			return token;
		}
	}

}
</span>

<span style="font-size:18px;">public class ThreadPoolBasedPipeExample
{

	/**
	 * 主函数
	 * 
	 * @param args
	 *            void
	 * @author lihong 2016年4月26日 下午2:43:54
	 * @since v1.0
	 */
	public static void main(String[] args)
	{
		/*
		 * 创建线程池
		 */
		final ThreadPoolExecutor executorSerivce;
		executorSerivce = new ThreadPoolExecutor(1, Runtime.getRuntime().availableProcessors() * 2, 60, TimeUnit.MINUTES, new SynchronousQueue<Runnable>(), new ThreadPoolExecutor.CallerRunsPolicy());

		/*
		 * 创建管道线对象
		 */
		final SimplePipeline<String, String> pipeline = new SimplePipeline<String, String>();

		/*
		 * 创建第一条管道
		 */
		Pipe<String, String> pipe = new AbstractPipe<String, String>()
		{

			@Override
			protected String doProcess(String input) throws PipeException
			{
				String result = input + "->[pipe1," + Thread.currentThread().getName() + "]";
				System.out.println(result);
				return result;
			}
		};

		/*
		 * 将第一条管道加入线程池
		 */
		pipeline.addAsThreadPoolBasedPipe(pipe, executorSerivce);

		/*
		 * 创建第二条管道
		 */
		pipe = new AbstractPipe<String, String>()
		{

			@Override
			protected String doProcess(String input) throws PipeException
			{
				String result = input + "->[pipe2," + Thread.currentThread().getName() + "]";
				System.out.println(result);
				try
				{
					Thread.sleep(new Random().nextInt(100));
				}
				catch (InterruptedException e)
				{
					;
				}
				return result;
			}
		};

		/*
		 * 将第二条管道加入管道线
		 */
		pipeline.addAsThreadPoolBasedPipe(pipe, executorSerivce);

		/*
		 * 创建第三条管道
		 */
		pipe = new AbstractPipe<String, String>()
		{

			@Override
			protected String doProcess(String input) throws PipeException
			{
				String result = input + "->[pipe3," + Thread.currentThread().getName() + "]";
				;
				System.out.println(result);
				try
				{
					Thread.sleep(new Random().nextInt(200));
				}
				catch (InterruptedException e)
				{
					;
				}
				return result;
			}

		};

		/*
		 * 将第三条管道加入管道线
		 */
		pipeline.addAsThreadPoolBasedPipe(pipe, executorSerivce);

		/*
		 * 第四条
		 */
		pipe = new AbstractPipe<String, String>()
		{

			@Override
			protected String doProcess(String input) throws PipeException
			{
				String result = input + "->[pipe4," + Thread.currentThread().getName() + "]";
				;
				System.out.println(result);
				try
				{
					Thread.sleep(new Random().nextInt(200));
				}
				catch (InterruptedException e)
				{
					;
				}
				return result;
			}

		};

		/*
		 * 将第四条管道加入管道线
		 */
		pipeline.addAsThreadPoolBasedPipe(pipe, executorSerivce);

		/*
		 * 创建第五条
		 */
		pipe = new AbstractPipe<String, String>()
		{

			@Override
			protected String doProcess(String input) throws PipeException
			{
				String result = input + "->[pipe5," + Thread.currentThread().getName() + "]";
				;
				System.out.println(result);
				try
				{
					Thread.sleep(new Random().nextInt(200));
				}
				catch (InterruptedException e)
				{
					;
				}
				return result;
			}

			@Override
			public void shutdown(long timeout, TimeUnit unit)
			{

				// 在最后一个Pipe中关闭线程池
				executorSerivce.shutdown();
				try
				{
					executorSerivce.awaitTermination(timeout, unit);
				}
				catch (InterruptedException e)
				{
					;
				}
			}
		};

		/*
		 * 将第五条管道加入管道线
		 */
		pipeline.addAsThreadPoolBasedPipe(pipe, executorSerivce);

		/*
		 * 管道线初始化
		 */
		pipeline.init(pipeline.newDefaultPipelineContext());

		int N = 100;
		try
		{
			for (int i = 0; i < N; i++)
			{
				pipeline.process("Task-" + i);
			}
		}
		catch (IllegalStateException e)
		{
			;
		}
		catch (InterruptedException e)
		{
			;
		}

		pipeline.shutdown(10, TimeUnit.SECONDS);

	}

}
</span>

<span style="font-size:18px;">Task-4->[pipe1,pool-1-thread-5]
Task-4->[pipe1,pool-1-thread-5]->[pipe2,pool-1-thread-4]
Task-4->[pipe1,pool-1-thread-5]->[pipe2,pool-1-thread-4]->[pipe3,pool-1-thread-3]
Task-4->[pipe1,pool-1-thread-5]->[pipe2,pool-1-thread-4]->[pipe3,pool-1-thread-3]->[pipe4,pool-1-thread-6]
Task-4->[pipe1,pool-1-thread-5]->[pipe2,pool-1-thread-4]->[pipe3,pool-1-thread-3]->[pipe4,pool-1-thread-6]->[pipe5,pool-1-thread-1]</span>

Java多线程编程实战指南（设计模式篇，黄文海）-之管道线模式

标签：

原文地址：http://blog.csdn.net/nmgrd/article/details/51258884