并发（14）：新类库中的构件（上）

落日映苍穹つ 2023-02-11 15:30 57阅读 0赞

java SE5的java.util.concurrent引入了大量设计用来解决并发问题的新类。学习使用它们将有助于你编写出更加简单而健壮的并发程序。

本节包含了各种组件具有代表性的示例，但是少数组件，即那些你不太可能用到或碰到的组件，没有包括在内。

因为这些组件设计各种问题，所以没有一种清晰的方式可以用来组织他们，因此我尝试着从最简单的示例入手，逐渐增加复杂度，从而介绍所有的示例。

### 一、CountDownLatch ###

它被用来同步一个或多个任务，强制它们等待由其他任务执行的一组操作完成。

你可以向CountDownLatch对象设置一个初始计数值，任何在这个对象上调用wait()的方法都将阻塞，直至这个计数值到达0。其他任务在结束其工作时，可以在该对象上调用countDown()来减小这个计数值。CountDownLatch被设计为只触发一次，计数值不能被重置。如果你需要能够重置计数值的版本，则可以使用CyclicBarrier。

调用countDown()的任务在产生这个调用时并没有被阻塞，只有对await()的调用会被阻塞，直至计数值到达0。

CountDownLatch的典型用法是将一个程序分为n个互相独立的可解决任务，并创建值为非0的CountDownLatch。当每个任务完成时，都会在这个锁存器上调用countDown()。等待问题被解决的任务在这个锁存器上调用await()，将它们自己拦住，直至锁存器计数结束。下面是演示这种技术的一个框架示例：

package concurrency;
    
    import java.util.Random;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 一部分任务
     */
    class TaskPortion implements Runnable {
    
    	private static int counter = 0;
    	private final int id = counter++;
    	private static Random r = new Random();
    	private final CountDownLatch latch;
    
    	TaskPortion(CountDownLatch latch) {
    		this.latch = latch;
    	}
    
    	@Override
    	public void run() {
    		try {
    			doWork();
    			latch.countDown();
    		} catch (InterruptedException e) {
    			// 可接受的退出方式
    		}
    	}
    
    	public void doWork() throws InterruptedException {
    		TimeUnit.MILLISECONDS.sleep(r.nextInt(2000));
    		System.out.println(this + "completed");
    	}
    
    	public String toString() {
    		return String.format("%1$-3d", id);
    	}
    }
    
    /**
     * 等待倒计时
     */
    class WaitingTask implements Runnable {
    
    	private static int counter = 0;
    	private final int id = counter++;
    	private final CountDownLatch latch;
    
    	WaitingTask(CountDownLatch latch) {
    		this.latch = latch;
    	}
    
    	@Override
    	public void run() {
    		try {
    			latch.await();
    			System.out.println("Latch barrier passed for " + this);
    		} catch (InterruptedException e) {
    			System.out.println(this + " interrupted");
    		}
    	}
    
    	public String toString() {
    		return String.format("WaitingTask %1$-3d ", id);
    	}
    }
    
    public class CountDownLatchDemo {
    	static final int SIZE = 100;
    
    	public static void main(String[] args) {
    		ExecutorService exec = Executors.newCachedThreadPool();
    		// 所有对象必须共享一个CountDownLatch对象
    		CountDownLatch latch = new CountDownLatch(SIZE);
    		for (int i = 0; i < 10; i++)
    			exec.execute(new WaitingTask(latch));
    		for (int i = 0; i < SIZE; i++)
    			exec.execute(new TaskPortion(latch));
    		System.out.println("启动所有任务");
    		exec.shutdown();// 所有任务完成后退出
    	}
    }

TaskPortion将随机地休眠一段时间，以模拟这部分工作的完成，而WaitingTask表示系统中必须等待的部分，它要等待到问题的初始化部分完成为止。所有任务都用了在main()中定义的同一个单一的CountDownLatch。

**（1）类库的线程安全**

注意，TaskPortion包含一个静态的Random对象，这意味着多个任何可能会同时调用Random.nextInt()。这是否安全呢？

如果存在问题，在这种情况下，可以通过向TaskPortion提供其自己的Random对象来解决。也就是说，通过移除static限定符的方式解决。但是这个问题对于java标准类库中的方法来说，也大都存在：哪些是线程安全的？哪些不是？

遗憾的是，JDK文档并没有指出这一点。Random.nextInt()碰巧是安全的，但是，你必须通过使用Web引擎，或者审视java类库代码，去逐个地揭示这一点。这对于被设计为支持，至少理论上支持并发的程序设计语言来说，并非是一件好事。

### 二、CyclicBarrier ###

CyclicBarrier适用于这样的情况：你希望创建一组任务，它们并行地执行工作，然后在进行下一个步骤之前等待，直至所有任务都完成（看起来有些像join()）。它使得所有的并行任务都将在栅栏处列队，因此可以一致地向前移动。这非常像CountDownLatch，只是CountDownLatch是只触发一次事件，而CyclicBarrier可以多次重用。

下面是一个赛马游戏示例，其中使用了CyclicBarrier：

package concurrency;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 马
     */
    class Horse implements Runnable {
    	private static int counter = 0;
    	private final int id = counter++;
    	// 迈步
    	private int strides = 0;
    	private static Random r = new Random();
    	private static CyclicBarrier barrier;
    
    	public Horse(CyclicBarrier b) {
    		barrier = b;
    	}
    
    	public synchronized int getStrides() {
    		return strides;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				synchronized (this) {
    					strides += r.nextInt(3);
    				}
    				barrier.await();
    			}
    		} catch (InterruptedException e) {
    			// 合法的退出方式
    		} catch (BrokenBarrierException e) {
    			// 这是我们想知道的
    			throw new RuntimeException(e);
    		}
    	}
    
    	public String toString() {
    		return "Horse " + id + " ";
    	}
    
    	public String tracks() {
    		StringBuilder s = new StringBuilder();
    		for (int i = 0; i < getStrides(); i++)
    			s.append("*");
    		s.append(id);
    		return s.toString();
    	}
    }
    
    /**
     * 赛马游戏
     */
    public class HorseRace {
    	static final int FINISH_LINE = 75;
    	private List<Horse> horses = new ArrayList<>();
    	private ExecutorService exec = Executors.newCachedThreadPool();
    	private CyclicBarrier barrier;
    
    	public HorseRace(int nHorses, final int pause) {
    		barrier = new CyclicBarrier(nHorses, () -> {
    			StringBuilder s = new StringBuilder();
    			for (int i = 0; i < FINISH_LINE; i++)
    				s.append("=");// 跑道上的栅栏
    			System.out.println(s);
    			for (Horse horse : horses)
    				System.out.println(horse.tracks());
    			for (Horse horse : horses)
    				if (horse.getStrides() >= FINISH_LINE) {
    					System.out.println(horse + "Won!");
    					exec.shutdownNow();
    					return;
    				}
    			try {
    				TimeUnit.MILLISECONDS.sleep(pause);
    			} catch (InterruptedException e) {
    				System.out.println("barrier-action sleep interrupted");
    			}
    		});
    		for (int i = 0; i < nHorses; i++) {
    			Horse horse = new Horse(barrier);
    			horses.add(horse);
    			exec.execute(horse);
    		}
    	}
    
    	public static void main(String[] args) {
    		int nHorses = 7;
    		int pause = 200;
    		if (args.length > 0) {
    			int n = new Integer(args[0]);
    			nHorses = n > 0 ? n : nHorses;
    		}
    		if (args.length > 1) {
    			int p = new Integer(args[1]);
    			pause = p > -1 ? p : pause;
    		}
    		new HorseRace(nHorses, pause);
    	}
    }

可以向CyclicBarrier提供一个“栅栏动作”，它是一个Runnable，当计数值到达0时自动执行——这是CyclicBarrier和CountDownLatch之间的另一个区别。这里，栅栏动作是作为匿名内部类创建的，它被提交给了CyclicBarrier的构造器。

我试图让每匹马都打印自己，但是之后的显示顺序取决于任务管理器。CyclicBarrier使得每匹马都要执行为了向前移动所必需执行的所有工作，然后必须在栅栏处等待其他所有的马都准备完毕。当所有的马都向前移动时，CyclicBarrier将自动调用Runnable栅栏动作任务，按顺序显示马和终点线的位置。

一旦所有任务都越过了栅栏，它就会自动地为下一回合比赛做好准备。

为了展示这个非常简单的动画效果，你需要将控制台视窗尺寸调整为小到只有马时，才会展示出来。

### 三、DelayQueue ###

这是一个无界的BlockingQueue，用于放置实现了Delayed接口的对象，其中的对象只能在其到期时才能从队列中取走。这汇总队列是有序的，即头对象的延迟到期的时间最长。如果没有任何延迟到期，那么就不会有任何头元素，并且poll()将返回null（正因为这样，你不能将null放置到这种队列中）。

下面是一个示例，其中的Delayed对象自身就是任务，而DelayedTaskConsumer将最“紧急”的任务（到期时间最长的任务）从队列中取出，然后运行它。注意，这样DelayQueue就成为了优先级队列的一种变体：

package concurrency;
    
    import static java.util.concurrent.TimeUnit.MILLISECONDS;
    import static java.util.concurrent.TimeUnit.NANOSECONDS;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.Delayed;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 延迟的任务
     */
    class DelayedTask implements Runnable, Delayed {
    
    	private static int counter = 0;
    	private final int id = counter++;
    	private final int delta;
    	private final long trigger;
    	protected static List<DelayedTask> sequence = new ArrayList<>();
    
    	public DelayedTask(int delayInMilliseconds) {
    		delta = delayInMilliseconds;
    		trigger = System.nanoTime() + NANOSECONDS.convert(delta, MILLISECONDS);
    		sequence.add(this);
    	}
    
    	@Override
    	public int compareTo(Delayed o) {
    		DelayedTask that = (DelayedTask) o;
    		if (trigger < that.trigger)
    			return -1;
    		if (trigger > that.trigger)
    			return 1;
    		return 0;
    	}
    
    	@Override
    	public long getDelay(TimeUnit unit) {
    		return unit.convert(trigger - System.nanoTime(), NANOSECONDS);
    	}
    
    	@Override
    	public void run() {
    		System.out.print(this + " ");
    	}
    
    	public String toString() {
    		return String.format("[%1$-4d]", delta) + " task " + id;
    	}
    
    	public String summary() {
    		return "(" + id + ":" + delta + ")";
    	}
    
    	/**
    	 * 结束哨兵
    	 */
    	public static class EndSentinel extends DelayedTask {
    
    		private ExecutorService exec;
    
    		public EndSentinel(int delayInMilliseconds, ExecutorService e) {
    			super(delayInMilliseconds);
    			exec = e;
    		}
    
    		public void run() {
    			for (DelayedTask pt : sequence) {
    				System.out.print(pt.summary() + " ");
    			}
    			System.out.println();
    			System.out.println(this + " Calling shutdownNow()");
    			exec.shutdownNow();
    		}
    	}
    }
    
    /**
     * 延迟任务消费者
     */
    class DelayedTaskConsumer implements Runnable {
    
    	private DelayQueue<DelayedTask> q;
    
    	public DelayedTaskConsumer(DelayQueue<DelayedTask> q) {
    		this.q = q;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted())
    				q.take().run(); // 使用当前线程运行task
    		} catch (InterruptedException e) {
    			// Acceptable way to exit;
    		}
    		System.out.println("Finished DelayedTaskConsumer");
    	}
    }
    
    public class DelayQueueDemo {
    	public static void main(String[] args) {
    		Random r = new Random();
    		ExecutorService exec = Executors.newCachedThreadPool();
    		DelayQueue<DelayedTask> queue = new DelayQueue<DelayedTask>();
    		// 用具有随机延迟的任务填充
    		for (int i = 0; i < 20; i++)
    			queue.put(new DelayedTask(r.nextInt(5000)));
    		// 设定停止点
    		queue.add(new DelayedTask.EndSentinel(5000, exec));
    		exec.execute(new DelayedTaskConsumer(queue));
    	}
    }

DelayTask包含了一个称为sequence的List<DelayedTask>，它保存了任务被创建的顺序，因此我们可以看到排序是按照实际发生的顺序执行的。

Delayed接口有一个方法名为getDelay()，它可以用来告知延迟到期有多长时间，或者延迟在多长时间之前已经到期。这个方法将强制我们去使用TimeUnit类，因为这就是参数类型。这会产生一个非常方便的类，因为你可以很容易地转换单位而无需作任何声明。例如，delta的值是以毫秒为单位存储的，但是java SE5的方法System.nanoTime()产生的时间则是以纳秒为单位的。你可以转换delta的值，方法是声明它的单位以及你希望以什么单位来表示，就像下面这样：

NANOSECONDS.convert(delta, MILLISECONDS);

在getDelay()中，希望使用的单位是作为unit参数传递进来的，你使用它将当前时间与触发时间之间的差转换为调用者要求的单位，而无需知道这些单位是什么（这是策略设计模式的一个简单示例，在这种模式中，算法的一部分是作为参数传递进来的）。

为了排序，Delayed接口还继承了Comparable接口，因此必须实现compareTo()，使其可以产生合理的比较。toString()和summary()提供了输出格式化，而嵌套的EndSentinel类提供了一种关闭所有事物的途径，具体做法是将其放置为队列的最后一个元素。

注意，因为DelayedTaskConsummer自身是一个任务，所以它有自己的Thread，它可以使用这个线程来运行从队列中获取的所有任务。由于任务是按照队列优先级的顺序执行的，因此在本例中不需要启动任何单独的线程来运行DelayedTask。

从输出中可以看到，任何创建的顺序对执行顺序没有任何影响，任务时按照所期望的延迟顺序执行的。

**如果本文对您有很大的帮助，还请点赞关注一下。**