Java同步器

class Solver
{
	final int N; //矩阵行数
	final float[][] data;  //矩阵数据
	final CyclicBarrier barrier;

	//该类用于处理一行数据, 会在处理完成后调用barrier等待.
	class Worker implements Runnable
	{
		int myRow;

		public Worker(int rows)
		{
			this.myRow = rows;
		}

		public void run()
		{
			while (!isDone())
			{
				//计算该行数据
				processRow(myRow);

				try
				{
					barrier.await();
				}
				catch (InterruptedException ex)
				{
					return;
				}
				catch (BrokenBarrierException ex)
				{
					return;
				}
			}
		}
	}

	public Solver(float[][] matrix)
	{
		this.data = matrix;
		N = matrix.length;
		barrier = new CyclicBarrier(N, new Runnable{
			public void run()
			{
				//整合所有行计算线程的计算结果是否符合要求.
				//如果符合要求就将isDone()置为true.
				mergeRows(...);
			}
		});

		for (int i = 0; i < N; i++)
		{
			new Thread(new Worker(i)).start();
		}

		waitUntilDone();
	}
}

class Driver
{
	//启动信号
	private CountDownLatch startSignal = new CountDownLatch(1);
	//完成信号
	private CountDownLatch doneSignal = new CountDownLatch(N);

	void main() throws InterruptedException
	{
		for (int i = 0; i < N; i++)
		{
			new Thread(new Worker(startSignal, doneSignal)).start();
		}

		//其他准备工作
		doSomethingElse();
		startSignal.countDown();
		doSomethingElse();
		doneSignal.await(); //等待Worker对象全部完成.
	}
}

class Worker implements Runnable
{
	private final CountDownLatch startSignal;
	private fianl  CountDownLatch doneSignal;

	public Worker(CountDownLatch start, CountDownLatch done)
	{
		this.startSignal = start;
		this.doneSignal = done;
	}

	public void run()
	{
		try
		{
			startSignal.await(); //等待driver准备好
			doWork();
			doneSignal.countDown();
		}
		catch (InterruptedException ex)
		{...}
	}

	private void doWork()
	{....}
}

class Driver2
{
	void main() throws InterruptedException
	{
		CountDownLatch doneSignal = new CountDownLatch(N);
		Executor executor = ...;
		
		for (int i = 0; i < N; i++)
		{
			//执行第i部分
			executor.executor(new WorkerRunnable(doneSignal, i));
		}

		doneSignal.await();
		doSomethingElse(); //所有部分完成后的整合
	}
}

class WorkerRunnable implements Runnable
{
	private final CountDownLatch doneSignal;
	private final int i;

	public WorkerRunnable(CountDownLatch done, int i)
	{
		this.doneSignal = done;
		this.i = i;
	}

	public void run()
	{
		try
		{
			doWork(i);
			doneSignal.countDown();
		}
		catch (InterruptedException ex)
		{...}
	}

	private void doWork(int i)
	{...}
}

class FillAndEmpty
{
	Exchanger<DataBuffer> exchanger = new Exchanger<DataBuffer>();
	DataBuffer initialEmptyBuffer = ...; //初始的空缓冲区,向其中填充数据
	DataBuffer initialFullBuffer = ...; //初始的满缓冲区,消费其中数据

	class FillingLoop implements Runnable()
	{
		public void run()
		{
			DataBuffer currentBuffer = initialEmptyBuffer;
			try
			{
				while (currentBuffer != null)
				{
					//向缓冲区填充数据
					addToBuffer(currentBuffer);

					if (currentBuffer.isFull())
					{
						currentBuffer = exchanger.exchange(currentBuffer);
					}
				}
			}
			catch (InterruptedException ex)
			{....}			
		}
	}

	class EmptyingLoop implements Runnable
	{
		public void run()
		{
			DataBuffer currentBuffer = initialFullBuffer;

			try
			{
				while (currentBuffer != null)
				{
					//从缓冲区中取数据
					takeFromBuffer(currentBuffer);

					if (currentBuffer.isEmpty())
					{
						currentBuffer = exchanger.exchange(currentBuffer);
					}
				}
			}
			catch (InterruptedException ex)
			{....}
		}
	}

	void start()
	{
		new Thread(new FillingLoop()).start();
		new Thread(new EmptyingLoop()).start();
	}
}

class Pool
{
	private static final int MAX_AVAILABLE = 100;
	private final Semaphore available = new Semaphore(MAX_AVAILABLE, true);

	public Object getItem()
	{
		available.acquire();
		return getNextAvailableItem();
	}

	public void putItem(Object x)
	{
		//将object置为未使用, 如果该object之前被使用过那么需要释放许可
		if (markAsUnused(x))
		{
			available.release();
		}
	}

	protected Object[] items = ...; //whatever kind of items
	//使用中的item,其used=true. 否则为false.
	protected boolean[] used = new boolean[MAX_AVAILABLE];

	protected synchronized Object getNextAvailableItem()
	{
		for (int i = 0; i < MAX_AVAILABLE; i++)
		{
			if (!used[i])
			{
				return items[i];
			}
		}
		return null;
	}

	//被传入的对象之前被使用过返回true,否则返回false
	protected boolean markAsUnused(Object x)
	{
		for (int i = 0; i < MAX_AVAILABLE; i++)
		{
			if (items[i] == x)
			{
				if (used[i])
				{
					used[i] = false;
					return true;
				}
				else
					return false;
			}
		}
		reture false;
	}
}