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

曾经终败给现在 2023-02-12 12:29 12阅读 0赞

### 六、Semaphore ###

正常的锁（来自concurrent.locks或内建的synchronized锁）在任何时刻都只允许一个任务访问一项资源，而计数信号量允许n个任务同时访问这个资源。你还可以将信号量看作是在向外分发使用资源的“许可证”，尽管实际上没有使用任何许可证对象。

作为一个示例，请考虑对象池的概念，它管理着数量有限的对象，当要使用对象时可以签出它们，而在用户使用完毕时，可以将它们签回。这种功能可以被封装到一个泛型类中：

package concurrency;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.Semaphore;
    
    public class Pool<T> {
    	private int size;
    	private List<T> items = new ArrayList<>();
    	private volatile boolean[] checkedOut;
    	private Semaphore available;
    
    	public Pool(Class<T> classObject, int size) {
    		this.size = size;
    		checkedOut = new boolean[size];
    		available = new Semaphore(size, true);
    		// 带有可签出的对象的负载池
    		for (int i = 0; i < size; i++) {
    			try {
    				items.add(classObject.newInstance());
    			} catch (InstantiationException e) {
    				e.printStackTrace();
    			} catch (IllegalAccessException e) {
    				e.printStackTrace();
    			}
    		}
    	}
    
    	public T checkOut() throws InterruptedException {
    		available.acquire();
    		return getItem();
    	}
    
    	public void checkIn(T x) {
    		if (releaseItem(x))
    			available.release();
    	}
    
    	private synchronized T getItem() {
    		for (int i = 0; i < size; ++i) {
    			if (!checkedOut[i]) {
    				checkedOut[i] = true;
    				return items.get(i);
    			}
    		}
    		return null; // Semaphore 防止到达这里
    	}
    
    	private synchronized boolean releaseItem(T item) {
    		int index = items.indexOf(item);
    		if (index == -1)
    			return false; // Not in the list
    		if (checkedOut[index]) {
    			checkedOut[index] = false;
    			return true;
    		}
    		return false;// 没有检出
    	}
    }

在这个简化的形式中，构造器使用newInstance()来把对象加载到池中。如果你需要一个新对象，那么可以调用checkOut()，并且在使用完之后，将其递交给checkIn()。

boolean类型的数组checkOut可以跟踪被签出的对象，并且可以通过getItem()和releaseItem()方法来管理。而这些都将由Semaphore类型的available来加以确保，因此，在checkOut()中，如果没有任何信号量许可证可用（这意味着池中没有更多的对象了），available将阻塞调用过程。在checkIn()中，如果被签入的对象有效，则会向信号量返回一个许可证。

为了创建一个示例，我们可以使用Fat，这是一种创建代价高昂的对象类型，因为它的构造器运行起来很耗时：

package concurrency;
    
    public class Fat {
    	private volatile double d;// 防止优化
    	private static int counter = 0;
    	private final int id = counter++;
    
    	public Fat() {
    		// 昂贵的, interruptible operation
    		for (int i = 0; i < 10000; i++) {
    			d += (Math.PI + Math.E) / (double) i;
    		}
    	}
    
    	public void operation() {
    		System.out.println(this);
    	}
    
    	public String toString() {
    		return "Fat id: " + id;
    	}
    }

我们在池中管理这些对象，以限制这个构造器所造成的影响。我们可以创建一个任务，它将签出Fat对象，持有一段时间之后再将它们签入，以此来测试Pool这个类：

package concurrency;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.Future;
    import java.util.concurrent.TimeUnit;
    
    /**
     * 从池中检出资源的任务
     */
    class CheckoutTask<T> implements Runnable {
    
    	private static int counter = 0;
    	private final int id = counter++;
    	private Pool<T> pool;
    
    	public CheckoutTask(Pool<T> pool) {
    		this.pool = pool;
    	}
    
    	@Override
    	public void run() {
    		try {
    			T item = pool.checkOut();
    			System.out.println(this + " checked out " + item);
    			TimeUnit.SECONDS.sleep(1);
    			System.out.println(this + " checking in " + item);
    			pool.checkIn(item);
    		} catch (InterruptedException e) {
    		}
    	}
    
    	public String toString() {
    		return "CheckoutTask " + id + " ";
    	}
    }
    
    public class SemaphoreDemo {
    	final static int SIZE = 25;
    
    	public static void main(String[] args) throws InterruptedException {
    		final Pool<Fat> pool = new Pool<>(Fat.class, SIZE);
    		ExecutorService exec = Executors.newCachedThreadPool();
    		for (int i = 0; i < SIZE; i++) {
    			exec.execute(new CheckoutTask<Fat>(pool));
    		}
    		System.out.println("All CheckoutTasks created");
    		List<Fat> list = new ArrayList<>();
    		for (int i = 0; i < SIZE; i++) {
    			Fat f = pool.checkOut();
    			System.out.print(i + ": main() thread checked out ");
    			f.operation();
    			list.add(f);
    		}
    		Future<?> blocked = exec.submit(() -> {
    			try {
    				// Semaphore 阻止额外的签出，因此调用被阻塞
    				pool.checkOut();
    			} catch (InterruptedException e) {
    				System.out.println("checkOut() Interrupted");
    			}
    		});
    		TimeUnit.SECONDS.sleep(2);
    		blocked.cancel(true);// Break out of blocked call
    		System.out.println("Checking in objects in " + list);
    		for (Fat f : list) {
    			pool.checkIn(f);
    		}
    		for (Fat f : list) {
    			pool.checkIn(f); // 第二次签入忽略
    		}
    		exec.shutdown();
    	}
    }

在main()中，创建了一个持有Fat对象的Pool，而一组CheckoutTask则开始操练这个Pool。然后，main()线程签出池中的Fat对象，但是并不签入它们。一旦池中所有的对象都被签出，Semaphore将不再允许执行任何签出操作。blocked的run()方法因此会被阻塞，2秒钟之后，cancel()方法被调用，以此来挣脱Future的束缚。注意，冗余的签入将被Pool忽略。

这个示例依赖于Pool的客户端严格地并愿意签入所持有的对象，当其工作时，这是最简单的解决方案。

### 七、Exchanger ###

Exchanger是在两个任务之间交换对象的栅栏。当这些任务进入栅栏时，它们各自拥有一个对象，当它们离开时，它们都拥有之前由对象持有的对象。Exchanger的典型应用场景是：一个任务在创建对象，这些对象的生产代价很高昂，而另一个任务在消费这些对象。通过这种方式，可以有更多的对象在被创建的同时被消费。

为了演练Exchanger类，我们将创建生产者和消费者任务，它们经由泛型和Generator，可以工作于任何类型的对象，然后我们将它们应用于Fat类。ExchangerProducer和ExchangerConsumer使用一个List<T>作为要交换的对象，它们都包含一个用于这个List<T>的Exchanger。当你调用Exchanger.exchanger()方法时，它将阻塞直至对方任务调用它自己的exchange()方法，那时，这两个exchange()方法将全部完成，而List<T>则被互换：

package concurrency;
    
    import java.util.List;
    import java.util.concurrent.CopyOnWriteArrayList;
    import java.util.concurrent.Exchanger;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    import com.buba.util.BasicGenerator;
    import com.buba.util.Generator;
    
    class ExchangerProducer<T> implements Runnable {
    
    	private Generator<T> generator;
    	private Exchanger<List<T>> exchanger;
    	private List<T> holder;
    
    	ExchangerProducer(Exchanger<List<T>> exchg, Generator<T> gen, List<T> holder) {
    		exchanger = exchg;
    		generator = gen;
    		this.holder = holder;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				for (int i = 0; i < ExchangerDemo.size; i++) {
    					holder.add(generator.next());
    				}
    				holder = exchanger.exchange(holder);
    			}
    		} catch (InterruptedException e) {
    		}
    	}
    }
    
    class ExchangerConsumer<T> implements Runnable {
    
    	private Exchanger<List<T>> exchanger;
    	private List<T> holder;
    	private volatile T value;
    
    	ExchangerConsumer(Exchanger<List<T>> ex, List<T> holder) {
    		exchanger = ex;
    		this.holder = holder;
    	}
    
    	@Override
    	public void run() {
    		try {
    			while (!Thread.interrupted()) {
    				holder = exchanger.exchange(holder);
    				for (T t : holder) {
    					value = t; // 拿出价值
    					holder.remove(t); // Ok for CopyOnWriteArrayList
    				}
    			}
    		} catch (InterruptedException e) {
    		}
    		System.out.println("Final value: " + value);
    	}
    }
    
    public class ExchangerDemo {
    	static int size = 10;
    	static int delay = 5; // Seconds
    
    	@SuppressWarnings({ "rawtypes", "unchecked" })
    	public static void main(String[] args) throws InterruptedException {
    		if (args.length > 0)
    			size = new Integer(args[0]);
    		if (args.length > 1)
    			delay = new Integer(args[1]);
    		ExecutorService exec = Executors.newCachedThreadPool();
    		Exchanger<List<Fat>> xc = new Exchanger<>();
    		List<Fat> producerList = new CopyOnWriteArrayList<Fat>(), consumerList = new CopyOnWriteArrayList<Fat>();
    		exec.execute(new ExchangerProducer(xc, BasicGenerator.create(Fat.class), producerList));
    		exec.execute(new ExchangerConsumer<Fat>(xc, consumerList));
    		TimeUnit.SECONDS.sleep(delay);
    		exec.shutdownNow();
    	}
    }

在main()中，创建了用于两个任务的单一的Exchanger，以及两个用于互换的CopyOnWriteArrayList。这个特定的List变体允许列表被遍历时调用remove()方法，而不会抛出ConcurrentModificationException异常。ExchangerProducer将填充这个List，然后将这个满列表交换为ExchangerConsumer传递给它的空列表。因为有了Exchanger，填充一个列表和消费另一个列表便可以同时发生了。

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