并发（12）：线程之间的协作（下）

曾经终败给现在 2023-02-11 07:19 1阅读 0赞

### 四、生产者-消费者与队列 ###

wait()和notifyAll()方法以一种非常低级的方式解决了任务互操作问题，即每次交互时都握手。在许多情况下，你可以瞄向更高的抽象级别，使用同步队列来解决任务协作问题，同步队列在任何时刻都只允许一个任务插入或移除元素。在java.util.concurrent.BlockingQueue接口中提供了这个队列，这个接口有大量的标准实现。你通常可以使用LinkedBlockingQueue，它是一个无界列队，还可以使用ArrayBlockQueue，它具有固定的尺寸，因此你可以在它被阻塞之前，向其中放置有限数量的元素。

如果消费者任务试图从队列中获取对象，而该对象此时为空，那么这些队列还可以挂起消费者任务，并且当有更多的元素可用时恢复消费者任务。阻塞队列可以解决非常大量的问题，而其方式与wait()和notifyAll()相比，则简单并可靠的多。

下面是一个简单的测试，它将多个LiftOff对象的执行串行化了。消费者是LiftOffRunner，它将每个LiftOff对象从BlockingQueue中推出并直接运行。（即，它通过显式地调用run()而使用自己的线程来运行，而不是为每个任务启动一个新线程。）

package concurrency;
    
    import java.io.BufferedReader;
    import java.io.InputStreamReader;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.SynchronousQueue;
    
    class LiftOffRunner implements Runnable {
    
    	private BlockingQueue<LiftOff> rockets;
    
    	public LiftOffRunner(BlockingQueue<LiftOff> queue) {
    		rockets = queue;
    	}
    
    	public void add(LiftOff lo) {
    		try {
    			rockets.put(lo);
    		} catch (InterruptedException e) {
    			System.out.println("Interrupted during put()");
    		}
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				LiftOff rocket = rockets.take();
    				rocket.run(); // Use this thread
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Waking from take()");
    		}
    		System.out.println("Exiting LiftOffRunner");
    	}
    }
    
    public class TestBlockingQueues {
    	static void getKey() {
    		try {
    			// compensate for Window/Linux difference in the
    			// length of the result produced by the Enter key
    			new BufferedReader(new InputStreamReader(System.in)).readLine();
    		} catch (java.io.IOException e) {
    			throw new RuntimeException(e);
    		}
    	}
    
    	static void getKey(String message) {
    		System.out.println(message);
    		getKey();
    	}
    
    	static void test(String msg, BlockingQueue<LiftOff> queue) {
    		System.out.println(msg);
    		LiftOffRunner runner = new LiftOffRunner(queue);
    		Thread t = new Thread(runner);
    		t.start();
    		for (int i = 0; i < 5; i++) {
    			runner.add(new LiftOff(5));
    		}
    		getKey("Press 'Enter' (" + msg + ")");
    		t.interrupt();
    		System.out.println("Finished " + msg + " test");
    	}
    
    	public static void main(String[] args) {
    		// Unlimited size
    		test("LinkedBlockingQueue", new LinkedBlockingQueue<LiftOff>());
    		// Fixed size
    		test("ArrayBlockingQueue", new ArrayBlockingQueue<LiftOff>(3));
    		// Size of 1
    		test("SynchronousQueue", new SynchronousQueue<LiftOff>());
    	}
    }

各个任务由main()放置到了BlockingQueue中，并且由LiftOffRunner从BlockingQueue中取出。注意，LiftOffRunner可以忽略同步问题，因为它们已经由BlockingQueue解决了。

**（1）吐司BlockingQueue**

考虑下面这个使用BlockingQueue的示例，有一台机器具有三个任务：一个制作吐司、一个给吐司抹黄油，另一个在抹过黄油的吐司上涂果酱。我们可以通过各个处理过程之间的BlockingQueue来运行这个吐司制作程序：

package concurrency;
    
    import java.util.Random;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 烤面包
     */
    class Toast {
    	public enum Status {
    		DRY, BUTTERED, JAMMED // 干燥, 抹过黄油, 涂过果酱
    	}
    
    	private Status status = Status.DRY;
    	private final int id;
    
    	public Toast(int idn) {
    		id = idn;
    	}
    
    	// 抹黄油
    	public void butter() {
    		status = Status.BUTTERED;
    	}
    
    	// 果酱
    	public void jam() {
    		status = Status.JAMMED;
    	}
    
    	public Status getStatus() {
    		return status;
    	}
    
    	public int getId() {
    		return id;
    	}
    
    	public String toString() {
    		return "Toast " + id + ": " + status;
    	}
    }
    
    class ToastQueue extends LinkedBlockingQueue<Toast> {
    }
    
    /**
     * 烤炉
     */
    class Toaster implements Runnable {
    
    	private ToastQueue toastQueue;
    	private int count = 0;
    	private Random r = new Random();
    
    	public Toaster(ToastQueue tq) {
    		toastQueue = tq;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				TimeUnit.MILLISECONDS.sleep(100 + r.nextInt(500));
    				// Make toast
    				Toast t = new Toast(count++);
    				System.out.println(t);
    				// Insert into queue
    				toastQueue.put(t);
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Toaster interrupted");
    		}
    		System.out.println("Toaster off");
    	}
    }
    
    /**
     * 在吐司上涂黄油
     */
    class Butterer implements Runnable {
    
    	private ToastQueue dryQueue, butteredQueue;
    
    	public Butterer(ToastQueue dry, ToastQueue buttered) {
    		dryQueue = dry;
    		butteredQueue = buttered;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				// 直到下一片吐司出炉
    				Toast t = dryQueue.take();
    				t.butter();
    				System.out.println(t);
    				butteredQueue.put(t);
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Butterer interrupted");
    		}
    		System.out.println("Butterer off");
    	}
    }
    
    /**
     * 把果酱涂在涂黄油的烤面包上
     */
    class Jammer implements Runnable {
    
    	private ToastQueue butteredQueue, finishedQueue;
    
    	public Jammer(ToastQueue buttered, ToastQueue finished) {
    		butteredQueue = buttered;
    		finishedQueue = finished;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				// 直到下一片吐司出炉
    				Toast t = butteredQueue.take();
    				t.jam();
    				System.out.println(t);
    				finishedQueue.put(t);
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Jammer interrupted");
    		}
    		System.out.println("Jammer off");
    	}
    }
    
    /**
     * 吃烤面包
     */
    class Eater implements Runnable {
    	private ToastQueue finishedQueue;
    	private int counter = 0;
    
    	public Eater(ToastQueue finished) {
    		finishedQueue = finished;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				Toast t = finishedQueue.take();
    				if (t.getId() != counter++ || t.getStatus() != Toast.Status.JAMMED) {
    					System.out.println(">>>> Error: " + t);
    					System.exit(1);
    				} else {
    					System.out.println("大声的咀嚼! " + t);
    				}
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Eater interrupted");
    		}
    		System.out.println("Eater off");
    	}
    }
    
    public class ToastOMatic {
    	public static void main(String[] args) throws InterruptedException {
    		ToastQueue dryQueue = new ToastQueue(), butteredQueue = new ToastQueue(), finishedQueue = new ToastQueue();
    		ExecutorService exec = Executors.newCachedThreadPool();
    		exec.execute(new Toaster(dryQueue));
    		exec.execute(new Butterer(dryQueue, butteredQueue));
    		exec.execute(new Jammer(butteredQueue, finishedQueue));
    		exec.execute(new Eater(finishedQueue));
    		TimeUnit.SECONDS.sleep(5);
    		exec.shutdownNow();
    	}
    }

Toast是一个使用enum值的优秀示例。注意，这个示例中没有任何显式的同步（即使用Lock对象或synchronized关键字同步），因为同步由队列（其内部是同步的）和系统的设计隐式地管理了——每片Toast在任何时刻都只由一个任务在操作。因为队列是阻塞，使得处理过程将被自动地挂起和恢复。你可以看到由BlockingQueue产生的简化十分明显。在使用显式的wait()和notifyAll()时存在的类和类之间的耦合被消除了，因为每个类都只和它的BlockingQueue通信。

### 五、任务间使用管道进行输入/输出 ###

通过输入/输出在线程间进行通信通常很有用。提供线程功能的类库以“管道”的形式对线程间的输入/输出提供了支持。它们在java输入/输出类库中的对应物就是PipedWriter类（允许任务向管道写）和PipedReader类（允许不同任务从同一个管道中读取）。这个模型可以看成是“生产者-消费者”问题的变体，这里的管道就是一个封装好的解决方案。管道基本上是一个阻塞队列，存在于多个引入BlockingQueue之前的版本中。

下面是一个简单的例子，两个任务使用一个管道进行通信：

package concurrency;
    
    import java.io.IOException;
    import java.io.PipedReader;
    import java.io.PipedWriter;
    import java.util.Random;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 发送器
     */
    class Sender implements Runnable {
    	private Random r = new Random();
    	private PipedWriter out = new PipedWriter();
    
    	public PipedWriter getPipedWriter() {
    		return out;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (true)
    				for (char c = 'A'; c <= 'z'; c++) {
    					out.write(c);
    					TimeUnit.MILLISECONDS.sleep(500);
    				}
    		} catch (IOException e) {
    			System.out.println(e + " Sender write exception");
    		} catch (InterruptedException e) {
    			System.out.println(e + " Sender sleep interrupted");
    		}
    	}
    }
    
    /**
     * 接收器
     */
    class Receiver implements Runnable {
    
    	private PipedReader in;
    
    	public Receiver(Sender sender) throws IOException {
    		in = new PipedReader(sender.getPipedWriter());
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (true)
    				// Blocks until characters are there
    				System.out.print("Read: " + (char) in.read() + ", ");
    		} catch (IOException e) {
    			System.out.println(e + "Receiver read exception");
    		}
    	}
    }
    
    public class PipedIO {
    	public static void main(String[] args) throws IOException, InterruptedException {
    		Sender sender = new Sender();
    		Receiver receiver = new Receiver(sender);
    		ExecutorService exec = Executors.newCachedThreadPool();
    		exec.execute(sender);
    		exec.execute(receiver);
    		TimeUnit.SECONDS.sleep(4);
    		exec.shutdownNow();
    	}
    }

Sender和Receiver代表了需要互相通信两个任务，Sender创建了一个PipedWriter，它是一个单独的对象；但是对于Receiver，PipedReader的建立必须在构造器中与一个PipedWriter相关联。Sender把数据放进Writer，然后休眠一段时间（随机数）。然而，Receiver没有sleep()和wait()。但当它调用read()时，如果没有更多的数据，管道将自动阻塞。

注意sender和receiver是在main()中启动的，即对象构造彻底完毕以后。如果你启动了一个没有构造完毕的对象，在不同的平台管道会产生不一致的行为（注意，BlockingQueue使用起来更加健壮而容易）。

在shutdownNow()被调用时，可以看到PipedReader与普通I/O之间最重要差异——PipedReader是可中断的。如果你将in.read()调用修改为System.in.read()，那么interrupt()将不能打断read()调用。

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