设计线程安全的类

在线程安全的程序中，虽然可以将程序的所有状态都保存在公有的静态域中，但与那些将状态封装起来的程序相比，这些程序的线程安全性更难以得到验证，并且咋修改时也更难以始终确保线程安全性。通过使用封装技术，可以使得在不对整个程序进行分析的情况下就可以判断一个类是否时线程安全的。一些基本的容器类并非时线程安全的，例如ArrayList和HashMap，可以使用Collections.synchronizedXX将这些线程不安全的类在多线程环境中安全的使用。

  1. Collections.synchronizedList(new ArrayList<>());
  2. Collections.synchronizedMap(new HashMap<>(2));

在设计线程安全类的过程中，需要包含以下三个基本要素：

  • 找出构成对象状态的所有变量
  • 找出约束状态变量的不变性条件

  • 建立对象状态的并发访问管理策略

    在现有的线程安全类中添加功能

    java类库中包含许多可以重用的基础模块，我们应该优先选择重用而不是创建新得类，重用能降低开发工作量，开发风险以及维护成本。只有当现有得类不能支持我们得操作时，我们才会考虑在不破坏线程安全性得情况下添加一个新得操作。
    例如，我们需要一个线程安全的链表，它需要提供一个原子的”若没有则添加”的操作。它的概念是在向容器中添加元素前，首先检查该元素是否已经存在，如果存在就不需要添加。由于这个类必须是线程安全的，那么这个”若没有则添加”的操作必须是原子的。如果不是，那么在某些执行情况下，同时有两个线程都检查这个元素没有存在，都执行了添加操作，那么这个容器中就包含了两个相同的元素。
    我们可以使用同步容器类来实现这个”若没有则添加”的操作，因Vector没有putIfAbsent方法，所以我们扩展了它

    1. public class MyVector<E> extends Vector<E> {
    3.   public synchronized boolean putIfAbsent(E e) {
    4.       boolean absent = !contains(e);
    5.       if (absent) {
    6.           add(e);
    7.       }
    8.       return absent;
    9.   }
    10. }

    看一下下面这段代码是否是线程安全的 ```java public class NotThreadSafeList {

    public List list = Collections.synchronizedList(new ArrayList());

    public synchronized boolean putIfAbsent(E e){

    1.   boolean absent = !list.contains(e);
    2.   if (absent) {
    3.       list.add(e);
    4.   }
    5.   return absent;

    }

}

  1. 其实这段代码不是线程安全的，虽然putIfAbsent已经声明被synchronized修饰了，但这个锁属于对象锁而不是list的锁，也就是说putIfAbsent相对于list的其他操作来说并不是原子性的，因此无法确保当putIfAbsent在执行时另外一个线程不会修改链表。<br />正确的使用方法是下面这种
  2. ```java
  3. public class ThreadSafeList<E> {
  5.     public List<E> list = Collections.synchronizedList(new ArrayList<E>());
  7.     public boolean putIfAbsent(E e) {
  8.         synchronized (list) {
  9.             boolean absent = !list.contains(e);
  10.             if (absent) {
  11.                 list.add(e);
  12.             }
  13.             return absent;
  14.         }
  15.     }
  16. }

这样就能确保putIfAbsent在对list进行操作时，其他线程不会修改list

并发容器

上述操作使用的容器属于同步容器，同步容器能将所有对容器的操作都串行化，以实现线程安全性，但是这种方法严重降低并发性，当多个线程竞争容器的锁时，系统吞吐量将严重降低。很幸运的是，在java5中提供了很多种并发容器类来改进同步容器的性能。通过并发容器来代替同步容器，可以极大地提高系统伸缩性并降低风险。

ConcurrentHashMap

与HashMap一样，ConcurrentHashMap也是一个基于散列的Map，但它使用了一种完全不同的加锁策略来提供更高的并发性和伸缩性。ConcurrentHashMap并不是将每个方法都在同一个锁上同步并使得每次只能有一个线程访问容器，而是使用一种粒度更细的加锁机制来实现更大程度的共享，这种机制称为分段锁。在这种机制中，执行读取操作的线程和执行写入操作的线程可以并发的访问Map，并且一定数量的写入线程可以并发修改Map。ConcurrentHashMap带来的结果是：在并发访问环境下将实现更高的吞吐量，而在单线程环境中只损失非常小的性能。
由于ConcurrentHashMap不能被加锁来执行独占访问，因此我们无法使用客户端加锁来创建新的原子操作，但是一些常用的复合操作，例如”若没有则添加”，”若相等则移除”，”若相等则替换”等都已经实现为原子操作并且在ConcurrentMap接口中声明，如下。如果你需要这样的功能，那么你就需要考虑使用ConcurrentMap了。

  1. public interface ConcurrentMap<K, V> extends Map<K, V> {
  2.     //若没有则添加
  3.     V putIfAbsent(K key, V value);
  4.     //若相等则移除
  5.     boolean remove(Object key, Object value);
  6.     //若相等则替换
  7.     boolean replace(K key, V oldValue, V newValue);
  8. }

CopyOnWriteArrayList

CopyOnWriteArrayList用于替代同步List，在某些情况下它提供了更好的并发性能，并在迭代期间不需要对容器进行加锁或复制。类似的CopyOnWriteArraySet用于替代同步Set。
“写入时复制”容器的线程安全性在于只要正确的发布一个事实不可变对象，那么在访问该对象时就不需要进一步的同步。在每次修改时，都会复制底层数组并重新发布一个新的容器副本，从而实现可变性，但这需要一定的开销，特别是当容器的规模较大时。因此仅当迭代操作远远多于修改操作时，才应该使用这个容器。

阻塞队列和生产者-消费者模式

阻塞队列提供了可阻塞的put和take方法，以及支持定时的offer和poll方法。如果队列已经满了，那么put方法会一直阻塞直到有空间可用；同理如果队列为空，那么take方法会一直阻塞直到有元素可用。
生产者-消费者模式能简化开发过程，因为它消除了生产者和消费者类之间的代码依赖性，此外还将生产数据和消费数据的过程解耦以简化工作负担。
阻塞队列支持生产者-消费者模式，在基于阻塞队列构建的生产者-消费者模式中，生产者不需要知道消费者的标识和数据，只需将数据放入队列即可；同理消费者也不需要知道生成者是谁，只需消费队列中的数据。
如果生产者生成数据的速率比消费者处理的速率快，那么数据会在队列中累积，最终耗尽内存。这个时候就需要考虑有界队列了，当队列充满时，生产者会一直阻塞，而消费者就有时间来赶上处理进度。阻塞队列提供了一个offer方法，如果数据不能被添加到队列中，那么将返回一个失败状态。这样开发者就能创建更多灵活的策略来处理负荷过载的情况了，比如说，将数据写入磁盘，减少生产者线程数量等等。
队列BlockingQueue的几种实现：

  • LinkedBlockingQueue 支持先进先出队列
  • ArrayBlockingQueue 支持先进先出队列
  • PriorityBlockingQueue 按优先级排序的队列，自定义排序队列
  • SynchronousQueue 没有数据缓冲的队列，相当于直接将生产者生产的数据直接交付于消费者处理

    例子

  1. //生产者模型
  2. public class Produce implements Runnable {
  4.     private final BlockingQueue<String> blockingQueue;
  6.     public Produce(BlockingQueue<String> blockingQueue) {
  7.         this.blockingQueue = blockingQueue;
  8.     }
  10.     @Override
  11.     public void run() {
  12.         try {
  13.             for (int i = 0; i < 10; i++) {
  14.                 System.out.println("生产第" + (i+1) + "件产品");
  15.                 blockingQueue.put((i+1) + "");
  16.                 Thread.sleep(1000);
  17.             }
  18.         } catch (InterruptedException e) {
  19.             Thread.currentThread().interrupt();
  20.         }
  21.     }
  22. }
  1. //消费者模型
  2. public class Consumer implements Runnable {
  4.     private final BlockingQueue<String> blockingQueue;
  6.     public Consumer(BlockingQueue<String> blockingQueue) {
  7.         this.blockingQueue = blockingQueue;
  8.     }
  10.     @Override
  11.     public void run() {
  12.         try {
  13.             while (true) {
  14.                 String take = blockingQueue.take();
  15.                 System.out.println("消费第"+take+"件产品");
  16.                 Thread.sleep(1000);
  17.             }
  18.         } catch (InterruptedException e) {
  19.            Thread.currentThread().interrupt();
  20.         }
  21.     }
  22. }
  1. //业务处理
  2. public class ProduceConsumer {
  3.     public static void main(String[] args) {
  4.         BlockingQueue<String> queue = new LinkedBlockingQueue<>();
  5.         Produce produce = new Produce(queue);
  6.         Consumer consumer = new Consumer(queue);
  8.         new Thread(produce).start();
  9.         new Thread(consumer).start();
  10.     }
  11. }

双端队列与工作密取

java6新增了两种容器类型，Deque和BlockingDeque，他们分别对Queue和BlockingQueue进行了扩展。Deque是是一个双端队列，实现了在队列头和队列尾的高效插入和移除。
image.png
如果一个消费者完成了自己双端队列中的全部工作，那么它可以从其他消费者双端队列末尾秘密的获取工作。这种模式叫工作密取，每个消费者都有各自的双端队列。这种模式比传统的生产者-消费者模式具有更高的可伸缩性，这是因为工作者线程不会在单个共享的任务队列上发生竞争。在大多数时候，他们都只是访问自己的双端队列，从而极大的减少了竞争。当工作者线程需要访问另一个队列时，他会从队列的尾部而不是从头部获取工作，因此进一步降低了队列的竞争程度。