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

亦凉 2023-02-12 07:24 41阅读 0赞

### 四、PriorityBlockingQueue ###

这是一个很基础的优先级队列，它具有可阻塞的读取操作。下面是一个示例，其中在优先级队列中的对象是按照优先级顺序从队列中出现的任务。PrioritizedTask被赋予了一个优先级数字，以此来提供这种顺序：

package concurrency;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.Queue;
    import java.util.Random;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.PriorityBlockingQueue;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 优先级任务
     */
    class PrioritizedTask implements Runnable, Comparable<PrioritizedTask> {
    
    	private Random r = new Random();
    	private static int counter = 0;
    	private final int id = counter++;
    	private final int priority;
    	protected static List<PrioritizedTask> sequence = new ArrayList<>();
    
    	public PrioritizedTask(int priority) {
    		this.priority = priority;
    		sequence.add(this);
    	}
    
    	@Override
    	public int compareTo(PrioritizedTask o) {
    		return priority < o.priority ? 1 : (priority > o.priority ? -1 : 0);
    	}
    
    	@Override
    	public void run() {
    		try {
    			TimeUnit.MILLISECONDS.sleep(r.nextInt(250));
    		} catch (InterruptedException e) {
    			// Acceptable way to exit
    		}
    		System.out.println(this);
    	}
    
    	public String toString() {
    		return String.format("[%1$-3d]", priority) + " Task " + id;
    	}
    
    	public String summary() {
    		return "(" + id + ":" + priority + ")";
    	}
    
    	/**
    	 * 结束哨兵
    	 */
    	public static class EndSentinel extends PrioritizedTask {
    		private ExecutorService exec;
    
    		public EndSentinel(ExecutorService e) {
    			super(-1); // 这个程序的最低优先级
    			exec = e;
    		}
    
    		public void run() {
    			int count = 0;
    			for (PrioritizedTask pt : sequence) {
    				System.out.print(pt.summary());
    				if (++count % 5 == 0)
    					System.out.println();
    			}
    			System.out.println();
    			System.out.println(this + " Calling shutdownNow()");
    			exec.shutdownNow();
    		}
    	}
    }
    
    /**
     * 优先级任务生成器
     */
    class PrioritizedTaskProducer implements Runnable {
    
    	private Random r = new Random();
    	private Queue<Runnable> queue;
    	private ExecutorService exec;
    
    	public PrioritizedTaskProducer(Queue<Runnable> q, ExecutorService e) {
    		queue = q;
    		exec = e; // Used for EndSentinel
    	}
    
    	@Override
    	public void run() {
    		// Unbounded queue; never blocks
    		// Fill it up fast with random priorities:
    		for (int i = 0; i < 20; i++) {
    			queue.add(new PrioritizedTask(r.nextInt(10)));
    			Thread.yield();
    		}
    		// 逐步引入最优先的工作
    		try {
    			for (int i = 0; i < 10; i++) {
    				TimeUnit.MILLISECONDS.sleep(250);
    				queue.add(new PrioritizedTask(10));
    			}
    			// 增加工作，最低优先级优先
    			for (int i = 0; i < 10; i++) {
    				queue.add(new PrioritizedTask(i));
    			}
    			// 一个哨兵来停止所有的任务
    			queue.add(new PrioritizedTask.EndSentinel(exec));
    		} catch (InterruptedException e) {
    		}
    		System.out.println("Finished PrioritizedTaskProducer");
    	}
    }
    
    /**
     * 消费者优先任务
     */
    class PrioritizedTaskConsumer implements Runnable {
    
    	private PriorityBlockingQueue<Runnable> q;
    
    	public PrioritizedTaskConsumer(PriorityBlockingQueue<Runnable> q) {
    		this.q = q;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted())
    				// Use current thread to run the task
    				q.take().run();
    		} catch (InterruptedException e) {
    		}
    		System.out.println("Finished PrioritizedTaskConsumer");
    	}
    }
    
    public class PriorityBlockingQueueDemo {
    	public static void main(String[] args) {
    		Random r = new Random();
    		ExecutorService exec = Executors.newCachedThreadPool();
    		PriorityBlockingQueue<Runnable> queue = new PriorityBlockingQueue<>();
    		exec.execute(new PrioritizedTaskProducer(queue, exec));
    		exec.execute(new PrioritizedTaskConsumer(queue));
    	}
    }

与前一个示例相同，PrioritizedTask对象的创建序列被记录在sequence List中，用于和实际的执行顺序比较。run()方法将休眠一小段随机的时间，然后打印对象信息，而EndSentinel提供了和前面相同的功能，要确保它是队列中最后一个对象。

PrioritizedTaskProducer和PrioritizedTaskConsumer通过PriorityBlockingQueue彼此连接。因此这种队列的阻塞特性提供了所有必需的同步，所以你应该注意到了，这里不需要任何显式的同步——不必考虑当你从这种队列中读取时，其中是否有元素，因为这个队列在没有元素时，将直接阻塞读取者。

### 五、使用ScheduledExecutor的温室控制器 ###

在内部类章节中介绍过可以应用于假象温室的控制系统的示例，它可以控制各种设施的开关，或者是对它们进行调节。这可以被看作是一种并发问题，每个期望的温室事件都是一个在预定时间运行的任务。ScheduledThreadPoolExecutor提供了解决该问题的服务。通过使用schedule()（运行一次任务）或者scheduleAtFixedRate()（每隔规则的时间重复执行任务），你可以将Runnable对象设置为将来的某个时刻执行。将下面的程序与在内部类章节中使用的方式相比，就会注意到，当你使用像ScheduledThreadPoolExecutor这样的预定义工具时，要简单许多：

package concurrency;
    
    import java.text.DateFormat;
    import java.util.ArrayList;
    import java.util.Calendar;
    import java.util.Collections;
    import java.util.List;
    import java.util.Random;
    import java.util.concurrent.ScheduledThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 温室调度器
     */
    public class GreenhouseScheduler {
    	// 灯
    	private volatile boolean light = false;
    	// 水
    	private volatile boolean water = false;
    	// 恒温器
    	private String thermostat = "Day";
    
    	public synchronized String getThermostat() {
    		return thermostat;
    	}
    
    	public synchronized void setThermostat(String value) {
    		thermostat = value;
    	}
    
    	ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(10);
    
    	// 计划
    	public void schedule(Runnable event, long delay) {
    		scheduler.schedule(event, delay, TimeUnit.MILLISECONDS);
    	}
    
    	// 重复
    	public void repeat(Runnable event, long initialDelay, long period) {
    		scheduler.scheduleAtFixedRate(event, initialDelay, period, TimeUnit.MILLISECONDS);
    	}
    
    	class LightOn implements Runnable {
    		@Override
    		public void run() {
    			// put hardware control code here to physically turn on the light
    			System.out.println("打开灯");
    			light = true;
    		}
    	}
    
    	class LightOff implements Runnable {
    		@Override
    		public void run() {
    			// put hardware control code here to physically turn off the light
    			System.out.println("打开灯");
    			light = false;
    		}
    	}
    
    	class WaterOn implements Runnable {
    		@Override
    		public void run() {
    			// put hardware control code here
    			System.out.println("打开温室的水");
    			water = true;
    		}
    	}
    
    	class WaterOff implements Runnable {
    		@Override
    		public void run() {
    			// put hardware control code here
    			System.out.println("关闭温室的水");
    			water = false;
    		}
    	}
    
    	/**
    	 * 恒温器 夜间
    	 */
    	class ThermostatNight implements Runnable {
    		@Override
    		public void run() {
    			System.out.println("把恒温器设置为夜间模式");
    			setThermostat("Night");
    		}
    	}
    
    	/**
    	 * 恒温器 白天
    	 */
    	class ThermostatDay implements Runnable {
    		@Override
    		public void run() {
    			System.out.println("把恒温器设置为白天模式");
    			setThermostat("Day");
    		}
    	}
    
    	/**
    	 * 铃声
    	 */
    	class Bell implements Runnable {
    		@Override
    		public void run() {
    			System.out.println("Bing!");
    		}
    	}
    
    	/**
    	 * 结束
    	 */
    	class Terminate implements Runnable {
    		@Override
    		public void run() {
    			System.out.println("Terminating");
    			scheduler.shutdownNow();
    			// 必须启动一个单独的任务来完成这项工作，因为调度程序已经关闭
    			new Thread() {
    				public void run() {
    					for (DataPoint d : data) {
    						System.out.println(d);
    					}
    				}
    			}.start();
    		}
    	}
    
    	/**
    	 * 新功能:数据收集 数据点
    	 */
    	static class DataPoint {
    		final Calendar time;
    		// 温度
    		final float temperature;
    		// 湿度
    		final float humidity;
    
    		public DataPoint(Calendar d, float temp, float hum) {
    			time = d;
    			temperature = temp;
    			humidity = hum;
    		}
    
    		public String toString() {
    
    			return DateFormat.getInstance().format(time.getTime())
    					+ String.format("温度：%1$.1f 湿度：%2$.2f", temperature, humidity);
    		}
    	}
    
    	private Calendar lastTime = Calendar.getInstance();
    
    	{
    		// 将日期调整到半小时
    		lastTime.set(Calendar.MINUTE, 30);
    		lastTime.set(Calendar.SECOND, 00);
    	}
    
    	// 上次温度
    	private float lastTemp = 65.0f;
    	// 温度的方向
    	private int tempDirection = +1;
    	// 上次湿度
    	private float lastHumidity = 50.0f;
    	// 湿度的方向
    	private int humidityDirection = +1;
    	private Random r = new Random();
    	List<DataPoint> data = Collections.synchronizedList(new ArrayList<>());
    
    	/**
    	 * 收集数据
    	 */
    	class CollectData implements Runnable {
    
    		@Override
    		public void run() {
    			System.out.println("正在收集数据");
    			synchronized (GreenhouseScheduler.this) {
    				// 假设间隔比实际时间长
    				lastTime.set(Calendar.MINUTE, lastTime.get(Calendar.MINUTE) + 30);
    				// 有五分之一的机会改变方向
    				if (r.nextInt(5) == 4)
    					tempDirection = -tempDirection;
    				// 存储以前的值
    				lastTemp = lastTemp + tempDirection * (1.0f + r.nextFloat());
    				if (r.nextInt(5) == 4)
    					humidityDirection = -humidityDirection;
    				// 存储以前的值
    				lastHumidity = lastHumidity + humidityDirection * r.nextFloat();
    				// 必须克隆Calendar,否则，所有DataPoint都持有对同一个lastTime的引用。
    				// 对于像calendar这样的基本对象，clone()是可以的
    				data.add(new DataPoint((Calendar) lastTime.clone(), lastTemp, lastHumidity));
    			}
    		}
    	}
    
    	public static void main(String[] args) {
    		GreenhouseScheduler gh = new GreenhouseScheduler();
    		gh.schedule(gh.new Terminate(), 5000);
    		// 以前的“重启”类没有必要
    		gh.repeat(gh.new Bell(), 0, 1000);
    		gh.repeat(gh.new ThermostatNight(), 0, 2000);
    		gh.repeat(gh.new LightOn(), 0, 200);
    		gh.repeat(gh.new LightOff(), 0, 400);
    		gh.repeat(gh.new WaterOn(), 0, 600);
    		gh.repeat(gh.new WaterOff(), 0, 800);
    		gh.repeat(gh.new ThermostatDay(), 0, 1400);
    		gh.repeat(gh.new CollectData(), 500, 500);
    	}
    }

这个版本重新组织了代码，并且添加了新的特性：收集温室内的温度和湿度度数。DataPoint可以持有并显示单个的数据段，而CollectData是被调度的任务，它在每次运行时，都可以产生仿真数据，并将其添加到Greenhouse的List<DataPoint>中。

注意，volatile和synchronized在适当的场合都得到了应用，以防止任务之间的互相干涉。在持有DataPoint的List中的所有方法都是synchronized的，这是因为在List被创建时，使用了java.util.Collection实用工具synchronizedList()。

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