并发（19）：仿真（下）

傷城~ 2023-02-13 11:52 17阅读 0赞

### 三、分发工作 ###

下面的仿真示例将本章的许多概念都结合在了一起。考虑一个假想的用于汽车的机器人组装线，每辆Car都将分多个阶段构建，从构建底盘开始，紧跟着是安装发动机、车厢和轮子。

package concurrency;
    
    import java.util.HashSet;
    import java.util.Set;
    import java.util.concurrent.BrokenBarrierException;
    import java.util.concurrent.CyclicBarrier;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 汽车
     */
    class Car {
    	private final int id;
    	// 引擎、动力传动系统、轮子
    	private boolean engine = false, driveTrain = false, wheels = false;
    
    	public Car(int idn) {
    		id = idn;
    	}
    
    	// Empty Car object
    	public Car() {
    		id = -1;
    	}
    
    	public synchronized int getId() {
    		return id;
    	}
    
    	public synchronized void addEngine() {
    		engine = true;
    	}
    
    	public synchronized void addDriveTrain() {
    		driveTrain = true;
    	}
    
    	public synchronized void addWheels() {
    		wheels = true;
    	}
    
    	@Override
    	public synchronized String toString() {
    		return "Car [id=" + id + ", engine=" + engine + ", driveTrain=" + driveTrain + ", wheels=" + wheels + "]";
    	}
    }
    
    /**
     * 汽车队列
     */
    class CarQueue extends LinkedBlockingQueue<Car> {
    }
    
    /**
     * 底盘构建器
     */
    class ChassisBuilder implements Runnable {
    
    	private CarQueue carQueue;
    	private int counter = 0;
    
    	public ChassisBuilder(CarQueue cq) {
    		carQueue = cq;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				TimeUnit.MILLISECONDS.sleep(500);
    				// 制作底盘
    				Car c = new Car(counter++);
    				System.out.println("ChassisBuilder created " + c);
    				// 添加到队列
    				carQueue.put(c);
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Interrupted: ChassisBuilder");
    		}
    		System.out.println("ChassisBuilder off");
    	}
    }
    
    /**
     * 装配程序
     */
    class Assembler implements Runnable {
    
    	private CarQueue chassisQueue, finishingQueue;
    	private Car car;
    	private CyclicBarrier barrier = new CyclicBarrier(4);
    	private RobotPool robotPool;
    
    	public Assembler(CarQueue cq, CarQueue fq, RobotPool rp) {
    		chassisQueue = cq;
    		finishingQueue = fq;
    		robotPool = rp;
    	}
    
    	public Car car() {
    		return car;
    	}
    
    	public CyclicBarrier barrier() {
    		return barrier;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				// 阻塞， 直到底盘是可用的
    				car = chassisQueue.take();
    				// 雇佣机器人来完成工作
    				robotPool.hire(EngineRobot.class, this);
    				robotPool.hire(DriveTrainRobot.class, this);
    				robotPool.hire(WheelRobot.class, this);
    				barrier.await();// 直到机器人完成
    				// 把车排好队，等待下一份工作
    				finishingQueue.put(car);
    			}
    		} catch (InterruptedException e) {
    			System.out.println("通过中断退出装配程序");
    		} catch (BrokenBarrierException e) {
    			// 这是我们想知道的
    			throw new RuntimeException(e);
    		}
    		System.out.println("Assembler off");
    	}
    }
    
    /**
     * 通信、报告
     */
    class Reporter implements Runnable {
    
    	private CarQueue carQueue;
    
    	public Reporter(CarQueue cq) {
    		carQueue = cq;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				System.out.println("报告： " + carQueue.take());
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Exiting Reporter via interrupt");
    		}
    		System.out.println("Reporter off");
    	}
    }
    
    /**
     * 机器人
     */
    abstract class Robot implements Runnable {
    	private RobotPool pool;
    
    	public Robot(RobotPool p) {
    		pool = p;
    	}
    
    	protected Assembler assembler;
    
    	public Robot assignAssembler(Assembler assembler) {
    		this.assembler = assembler;
    		return this;
    	}
    
    	private boolean engage = false;
    
    	// 参与
    	public synchronized void engage() {
    		engage = true;
    		notifyAll();
    	}
    
    	// run()的部分对于每个机器人都是不同的
    	abstract protected void performService();
    
    	public void run() {
    		try {
    			powerDown(); // 等待需要
    			while (!Thread.interrupted()) {
    				performService();
    				assembler.barrier().await(); // 同步
    				// 我们的工作结束了
    				powerDown();
    			}
    		} catch (InterruptedException e) {
    			System.out.println("Exiting " + this + " via interrupt");
    		} catch (BrokenBarrierException e) {
    			// 这是我们想知道的
    			throw new RuntimeException(e);
    		}
    		System.out.println(this + " off");
    	}
    
    	private synchronized void powerDown() throws InterruptedException {
    		engage = false;
    		assembler = null; // 断开与装配程序的连接
    		pool.release(this);
    		while (engage == false) // Power down
    			wait();
    	}
    
    	public String toString() {
    		return getClass().getName();
    	}
    }
    
    /**
     * 引擎的机器人
     */
    class EngineRobot extends Robot {
    
    	public EngineRobot(RobotPool pool) {
    		super(pool);
    	}
    
    	@Override
    	protected void performService() {
    		System.out.println(this + " 安装引擎");
    		assembler.car().addEngine();
    	}
    }
    
    /**
     * 传动系的机器人
     */
    class DriveTrainRobot extends Robot {
    
    	public DriveTrainRobot(RobotPool p) {
    		super(p);
    	}
    
    	@Override
    	protected void performService() {
    		System.out.println(this + " 安装动力传动系统");
    		assembler.car().addDriveTrain();
    	}
    }
    
    /**
     * 车轮机器人
     */
    class WheelRobot extends Robot {
    
    	public WheelRobot(RobotPool p) {
    		super(p);
    	}
    
    	@Override
    	protected void performService() {
    		System.out.println(this + " 安装轮胎");
    		assembler.car().addWheels();
    	}
    }
    
    /**
     * 机器人池
     */
    class RobotPool {
    	// 悄悄地防止相同的条目
    	private Set<Robot> pool = new HashSet<>();
    
    	public synchronized void add(Robot r) {
    		pool.add(r);
    		notifyAll();
    	}
    
    	public synchronized void hire(Class<? extends Robot> robotType, Assembler d) throws InterruptedException {
    		for (Robot r : pool) {
    			if (r.getClass().equals(robotType)) {
    				pool.remove(r);
    				r.assignAssembler(d);
    				r.engage();// 启动电源来完成任务
    				return;
    			}
    		}
    		wait();// 没有可用的
    		hire(robotType, d);// 再试一次,递归地
    	}
    
    	public synchronized void release(Robot r) {
    		add(r);
    	}
    }
    
    /**
     * 汽车制造者
     */
    public class CarBuilder {
    	public static void main(String[] args) throws InterruptedException {
    		CarQueue chassisQueue = new CarQueue(), finishingQueue = new CarQueue();
    		ExecutorService exec = Executors.newCachedThreadPool();
    		RobotPool robotPool = new RobotPool();
    		exec.execute(new EngineRobot(robotPool));
    		exec.execute(new DriveTrainRobot(robotPool));
    		exec.execute(new WheelRobot(robotPool));
    		exec.execute(new Assembler(chassisQueue, finishingQueue, robotPool));
    		exec.execute(new Reporter(finishingQueue));
    		// 开始一切运行生产底盘
    		exec.execute(new ChassisBuilder(chassisQueue));
    		TimeUnit.SECONDS.sleep(7);
    		exec.shutdownNow();
    	}
    }

Car是经由CarQueue从一个地方传送到另一个地方的，CarQueue是一种LinkedBlockingQueue类型。ChassisBuilder创建了一个未修饰的Car，并将它放到了一个CarQueue中。Assembler从一个CarQueue中取走Car，并雇请Robot对其加工。CyclicBarrier使Assembler等待，直至所有的Robot都完成，·并且在那一刻它将会将Car放置到即将离开它的CarQueue中，然后被传送到下一个操作。最终的CarQueue的消费者是一个Reporter对象，它只是打印Car，以显示所有的任务都已经正确的完成了。

Robot是在池中管理的，当需要完成工作时，就会从池中雇请适当的Robot。在工作完成时，这个Robot会返回到池中。

在main()中创建了所有必需的对象，并初始化了各个任务，最后启动ChassisBuilder，从而启动整个过程（但是，由于LinkedBlockingQueue的行为，使得最先启动它也没有问题）。注意，这个程序遵循了本章描述的所有有关对象和任务生命周期的设计原则，因此关闭这个过程将是安全的。

你会注意到，Car将其所有方法都设置成了synchronized的。正如它所表现出来的那样，在本例中，这是多余的，因为在工厂的内部，Car是通过队列移动的，并且在任何时刻，只有一个任务能够在某辆车上工作。基本上，队列可以强制串行化地访问Car。但是这正是你可能会落入的陷阱——你可能会说“让我们尝试着通过不对Car类同步来进行优化，因为看起来Car在这里并不需要同步。”但是稍后，当这个系统连接到另一个需要Car被同步的系统时，它就会崩溃。

Brian Goetz的注释：

进行这样的声明会简单得多：“Car可能会被多个线程使用，因此我们需要以明显的方式使其成为线程安全的。”我把这种方式描绘为：在公园中，你会在陡峭的坡路上发现一些保护围栏，并且可能会发现标记声明：“不要依靠围栏。”当然，这条规则的真实目的不是要阻止你借助围栏，而是防止你跌落悬崖。但是“不要倚靠围栏”与“不要跌落悬崖”相比，是一条遵循起来要容易得多的规则。

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