004 - 同步容器类

如果线程A在包含这么多元素的基础上调用getVector方法， 会得到一个数值， getVector只是取
得该元素， 而并不是从vector中移除， removeVector方法是得到一个元素进行移除， 这段代码的不
安全因素就是， 因为线程的时间片是乱序的， 而且getVector和remove Vector并不会保证互斥， 所以
在removeVector方法把某个值比如6666移除后， vector中就不存在这个6666的元素， 此时
getVector方法取得6666， 就会抛出数组越界异常。

为什么是数组越界异常呢?可以看一下vector的源码

如果用图表示的话，则会是下面这样

所以，从系统的层面来看，上面这段代码也要保证线程安全性才可以，也就是在客户端加锁实现，

只要让复合操作使用一把锁，那么这些操作就和其他单独的操作一样都是原子性的。

如下面例子所示

也可以通过锁住.class来保证原子性操作， 也能达到同样的效果。

在调用size和get之间， Vector的长度可能会发生变化， 这种变化在对Vector进行排序时出现， 如下所示

这种迭代的操作正确性取决于运气， 即在调用size和get之间会修改Vector， 在单线程环境中， 这种
假设完全成立， 但是再有其他线程并发修改Vector时， 则可能会导致麻烦。

仍旧可以通过客户端加锁的方式来避免这种情况

这种方式为客户端的可靠性提供了保证，但是牺牲了伸缩性，而且这种在遍历过程中进行加锁，也不是我们所希望看到的。

若没有则添加

使用ArrayList来举例子

该段代码会发生异常， 因为在ArrayList内部， 有两个属性

 ArrayList在remove等对集合结构的元素造成数量上的操作会有
checkForComodification的判断， 如下所示， 这也是这段代码的错误原因。

子主题

是ArrayList 中的一个成员变量。

它表示该集合实际被修改的次数。

List<String> userNames = new ArrayList<String>() {{
add("Hollis");
add("hollis");
add("HollisChuang");
add("H");
}};
当使用以上代码初始化集合之后该变量就有了。初始值为0。

是ArrayList 中的一个内部类——Itr 中的成员变量。

Iterator iterator = userNames.iterator();

以上代码，即可得到一个Itr 类，该类实现了Iterator 接口。

表示这个迭代器预期该集合被修改的次数。

其值随着Itr 被创建而初始化。

只有通过迭代器对集合进行操作，该值才会改变。

可以看到，它只修改了modCount，并没有对expectedModCount 做任何操作。

子主题

0表示解锁

其他所有的值表示加锁

比1大的值表示加锁的次数。

提供了一个等待队列(wait queue) 允许多个进程在锁上排队等待

除非内核明确的对他们解除阻塞，否则它们不会运行。

如果mutex没有加锁， 线程能够马上进入并且互斥量能够自动锁定， 从而阻止其他线程进入。

如果mutex已经加锁， 调用线程会阻塞，直到mutex解锁。

与互斥量有关的函数调用

和想象中的一样， mutex能够被创建和销毁， 扮演这两个角色的分别是

mutex也可以通过Pthread_mutex_Lock来进行加锁， 如果互斥量已经加锁， 则会阻塞调用者。

还有一个调用Pthread_mutex_try Lock用来尝试对线程加锁

最后， Pthread_mutex_unlock会对mutex解锁并且释放一个正在等待的线程。

条件变量(不像是信号量)不会存在于内存中。

如果将一个信号量，传递给一个没有线程等待的条件变量，那么这个信号就会丢失

下面进一步来研究线程、互斥量、条件变量之间的关联

一个线程将东西放在一个缓冲区内，由另一个线程将它们取出。

如果生产者发现缓冲区没有空槽可以使用了，生产者线程会阻塞起来直到有一个线程可以使用。

生产者使用mutex来进行原子性检查从而不受其他线程干扰。

但是当发现缓冲区已经满了以后， 生产者需要一种方法来阻塞自己并在以后被唤醒。

这便是条件变量做的工作。

与条件变量有关的重要的pthread函数调用

表中给出了一些调用用来创建和销毁条件变量。

条件变量上的主要属性是

Pthread_cond_broadcast用来唤醒多个阻塞的、需要等待信号唤醒的线程。

同时具有多个CPU的情况

每个CPU都有自己的私有内存

它们通过网络相连，那么这些原语将会失效，

所以还需要其他方法

为了防止消息丢失，发送方和接收方可以达成一致：

一旦接受到消息后， 接收方马上回送一条特殊的确认(acknowLedgement) 消息，

 如果发送方在一段时间间隙内未收到确认，则重发消息。

现在考虑消息本身被正确接收，而返回给发送着的确认消息丢失的情况。

发送者将重发消息，这样接受者将收到两次相同的消息。

图解

对于接收者来说，如何区分新的消息和一条重发的老消息是非常重要的。

通常采用在每条原始消息中嵌入一个连续的序号来解决此问题。

如果接受者收到一条消息，它具有与前面某一条消息一样的序号，就知道这条消息是重复的，可以忽略。

以便在发送或接收调用中清晰的指明进程。

比如客户端怎么知道它是在与一个真正的文件服务器通信

 从发送方到接收方的信息有可能被中间人所篡改。

子主题

ConcurrentHashMap继承了AbstractMap接口，并实现了ConcurrentMap和Seralizable接口

AbstractMap和ConcurrentMap都是Map的实现类， 只不过AbstractMap是抽象实现。

所有访问Hashtable的线程都想获取同一把锁

ConcurrentHashMap采用的分段锁实现

实现带来的结果是

 在并发环境下，可以实现更高的吞吐量

在单线程环境下，只损失非常小的性能

HashMap

ConcurrentHashMap

HashMap

ConcurrentHashMap

在线程安全的Map实现都实现了独占访问， 因此只能单个线程修改Map， 

ConcurrentHashMap

使用场景

实现方式

锁机制与线程安全

四个新的方法

这四个方法都是原子性方法， 进一步扩展了Map的功能

java.util.concurrent.Concurrent Navigable Map类是java.util.NavigableMap的子类

图解

它支持并发访问，并且允许其视图的并发访问

ConcurrentNavigableMap接口,包含一些可能有用的其他方法

线程安全的有序的哈希表

适用于高并发的场景

继承关系

底层数据结构是基于跳表实现的。

ConcurrentSkipListMap可以提供Comparable内部排序或者是Comparator外部排序， 具体取决于使用哪个构造函数。

线程安全的有序的集合

适用于高并发的场景

底层是通过ConcurrentNavigableMap来实现的， 它是一个有序的线程安全的集合。

有序的， 基于元素的自然排序或者通过比较器确定的顺序

线程安全的

CopyOnWriteArrayList就是ArrayList的一种线程安全的变体

每次并发修改，CopyOnWriteArrayList都相当于重新创建副本

其内部有一个指向数组的引用， 数组是不会被修改的

CopyOnWriteArrayList中的所有可变操作（比如add和set等）都是通过对数组进行全新复制来实现的。

这个过程存在一定的开销，所以，一般在规模很大时，读操作要远远多于写操作时， 为了保证线程安全性， 会使用Copy On Wrte ArrayList.

它就不会出现异常

一种fail-safe机制的，它不会抛出Concurrent Modification Exception异常， 

并且返回元素与迭代器创建时的元素相同。

比如CopyOnWriteArrayList案例

图解CopyOnWriteArraySet继承关系

ConcurrentLinkedQueue

PriorityQueue

阻塞队列，关键字是阻塞，先理解阻塞的含义，

译为阻塞队列

它是JDK 1.5添加的新的工具类

 它继承于Queue队列， 并扩展了Queue的功能

在检索元素时，会等待队列变成非空， 并在存储元素时，会等待队列变为可用

当队列中没有数据的情况下，消费者端的所有线程都会被自动阻塞（挂起），直到有数据放入队列。

图1 线程阻塞情况1

当队列中填满数据的情况下，生产者端的所有线程都会被自动阻塞（挂起），直到队列中有空的位置，线程被自动唤醒。

图2 线程阻塞情况2

BlockingQueue的继承关系

第一种是抛出异常

特殊值：第二种是根据情况会返回null或者false

阻塞：第三种是无限期的阻塞当前线程直到操作变为可用后

超时：第四种是给定一个最大的超时时间，超过后才会放弃

在其实现类的方法add、put或者offer后时，添加null会抛出空指针异常。

在任意时间内， 它都会有一个remainCapacity，

超过该值之前， 任意put元素都会阻塞。

如下图所示

阻塞队列的主要方法

方法类型

插入操作

获取数据操作：

阻塞队列家族

ArrayBlockingQueue

LinkedBlockingQueue

PriorityBlockingQueue

DelayQueue

SynchronousQueue

TransferQueue

LinkedTransferQueue

LinkedBlockingDeque

而双端队列适用于工作密取

分别实现了在队列头和队列尾的插入

ArrayDeque

ConcurrentLinkedDeque

LinkedBlockingDeque

继承关系：对Queue和BlockingQueue做了扩展

LinkedBlockingDeque

内存地址V

旧的预期值A

即将要更新的目标值B

CAS指令执行时，当且仅当内存地址V的值与预期值A相等时，将内存地址V的值修改为B，否则就什么都不做。

V表示要更新的变量(内存值)，E表示预期值(旧的)，N表示新值。

当且仅当V值等于E值时，才会将V的值设为N，

如果V值和E值不同，则说明已经有其他线程做了更新，则当前线程什么都不做。

最后，CAS返回当前V的真实值。

如果内存地址V初次读取的值是A，并且在准备赋值的时候检查到它的值仍然为A，但可能期间它已经被修改过了

通常将 CAS 用于同步的方式是从地址 V 读取值 A，执行多步计算来获得新值 B，然后使用 CAS 将 V 的值从 A 改为 B。

如果 V 处的值尚未同时更改，则 CAS 操作成功。

比如说一个线程one从内存位置V中取出A，这时候另一个线程two也从内存中取出A，并且two进行了一些操作变成了B，然后two又将V位置的数据变成A，

这时候线程one进行CAS操作发现内存中仍然是A，然后one操作成功。

尽管线程one的CAS操作成功，但是不代表这个过程就是没有问题的。

使用AtomicStampedReference的版本号机制来管理

部分乐观锁的实现是通过版本号（version）的方式

（通过state的get/set/CAS)

getState()

setState() 

compareAndSetState()

tryAcquire(int)/tryAcquireShared(int )

Exclusive独占资源-ReentrantLock

Exclusive（独占，只有一个线程能执行，如ReentrantLock）

tryRelease/tryReleaseShared(int)

Share共享资源-Semaphore/CountDownLatch

Share（共享，多个线程可同时执行，如Semaphore/CountDownLatch）。

一般来说，自定义同步器要么是独占方法，要么是共享方式，只需实现tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。

但AQS也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock。

既可独占，也可共享（ReentrantReadWriteLock）

什么是CAS（比较并交换-乐观锁机制-锁自旋）

实际应用

存在的问题

信号量

翻译过来就是信号量

它其实就是一种信号， 在操作系统中， 也有信号量的这个概念

用来控制同时访问特定资源的线程数量， 它通过协调各个线程，以保证合理的使用公共资源。

Semaphore管理着一组许可(permit) ， 许可的初始数量由构造函数来指定。

在获取某个资源之前，应该先从信号量获取许可(permit)， 以确保资源是否可用

当线程完成对资源的操作后， 会把它放在池中并向信号量返回一个许可，从而允许其他线程访问资源，这叫做释放许可。

如果没有许可的话，那么acquire将会阻塞直到有许可(中断或者操作超时) 为止。release方法将返回一个许可信号量，

例如常用的数据库连接池， 线程请求资源时， 

翻译成字面意思为 信号量

控制同时访问的线程个数

一种基于计数的信号量。

Semaphore可以用来构建一些对象池，资源池之类的，比如数据库连接池

实现互斥锁（计数器为1）

例子：若一个工厂有5台机器，但是有8个工人，一台机器同时只能被一个工人使用，只有使用完了，其他工人才能继续使用。通过Semaphore来实现

可以用于做流量控制，特别是公用资源有限的应用场景，比如数据库连接

acquire()

release()

public void acquire(): 用来获取一个许可，若无许可能够获得，则会一直等待，直到获得许可。

public void acquire(int permits):获取permits个许可

public void release() { } :释放许可。注意，在释放许可之前，必须先获获得许可。

public void release(int permits) { }:释放permits个许可

上面4个方法都会被阻塞，如果想立即得到执行结果，可以使用下面几个方法

public boolean tryAcquire():尝试获取一个许可，若获取成功，则立即返回true，若获取失败，则立即返回false

public boolean tryAcquire(long timeout, TimeUnit unit):尝试获取一个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false

public boolean tryAcquire(int permits):尝试获取permits个许可，若获取成功，则立即返回true，若获取失败，则立即返回false

public boolean tryAcquire(int permits, long timeout, TimeUnit unit): 尝试获取permits个许可，若在指定的时间内获取成功，则立即返回true，否则则立即返回false

availablePermits()方法得到可用的许可数目。

Semaphore基本能完成ReentrantLock的所有工作，使用方法也与之类似，通过acquire()与release()方法来获得和释放临界资源。

经实测，Semaphone.acquire()方法默认为可响应中断锁，与ReentrantLock.lockInterruptibly()作用效果一致，也就是说在等待临界资源的过程中可以被Thread.interrupt()方法中断。

此外，Semaphore也实现了可轮询的锁请求与定时锁的功能，除了方法名tryAcquire与tryLock不同，其使用方法与ReentrantLock几乎一致。

Semaphore也提供了公平与非公平锁的机制，也可在构造函数中进行设定。

Semaphore的锁释放操作也由手动进行，因此与ReentrantLock一样，为避免线程因抛出异常而无法正常释放锁的情况发生，释放锁的操作也必须在finally代码块中完成。

闭锁(Latch) 是一种同步工具类

它可以延迟线程的进度以直到其到达终止状态

闭锁的作用相当于是一扇门，

通过闭锁来启动一组相关的操作，使用闭锁来等待一组事件的执行。

闭锁是一种一次性对象，一旦进入终止状态后，就不能被置

闭锁的一种实现

它可以使一个或者多个线程等待一组事件的发生

闭锁有一个计数器， 闭锁需要对计数器进行初始化， 表示需要等待的次数

闭锁在调用await处进行等待， 其他线程在调用countDown把闭锁count次数进行递减， 直到递减为0， 唤醒await。

如下代码所示

ABCD四条线程， 必须同时到达Barrier， 然后手牵手一起走过幸福的殿堂。

countDown()

await()

开启多个线程分块下载一个大文件，每个线程只下载固定的一截，最后由另外一个线程来拼接所有的分段。

应用程序的主线程希望在负责启动框架服务的线程已经启动所有的框架服务之后再执行。

确保一个计算不会执行，直到所需要的资源被初始化。

public int await()：用来挂起当前线程，直至所有线程都到达barrier状态再同时执行后续任务；

public int await(long timeout, TimeUnit unit)：让这些线程等待至一定的时间，如果还有线程没有到达barrier状态就直接让到达barrier的线程执行后续任务。

多线程计算

而CyclicBarrier是可以重用的。

图解

segment

hashEntry

Node数组

链表

红黑树

流程

不用加锁，是非阻塞的

需要进行两次Hash操作，第一次Hash定位到Segment，第二次Hash定位到元素所在的链表的头部

定位到链表头部之后根据key取出对应的value值

因为HashEntry中的next是final的，一经赋值以后就不可修改，所以在定位到待删除元素e的位置以后，程序就将待删除元素前面的那一些元素全部复制一遍，然后再一个一个重新接到链表上去。尾结点指向e的下一个结点。e后面的结点不需要复制，它们可以重用。e之前的元素在remove()之后为remove之前的逆置

size()操作涉及到多个segment

size操作就是遍历了两次Segment，每次记录Segment的modCount值，然后将两次的modCount进行比较，如果相同，则表示期间没有发生过写入操作，就将原先遍历的结果返回，如果不相同，则把这个过程再重复做一次，如果再不相同，则就需要将所有的Segment都锁住，然后一个一个遍历了

它的add/remove 等方法都已经加锁了，还要copy 一份再修改干嘛？多此一举？

同样是线程安全的集合，这玩意和Vector 有啥区别呢？

Copy-On-Write

简称COW

是一种用于程序设计中的优化策略

从一开始大家都在共享同一个内容，当某个人想要修改这个内容时，才会真正把内容Copy 出去形成一个新的内容然后再改，

这是一种延时懒惰策略。

CopyOnWrite 容器即写时复制的容器。

CopyOnWrite 容器非常有用，可以在非常多的并发场景中使用到。

通俗的理解是

当往一个容器添加元素时，

适合多读少写的场景

CopyOnWrite 并发容器用于读多写少的并发场景。

比如白名单，黑名单，商品类目的访问和更新场景。

CopyOnWriteArrayList 中add/remove 等写方法是需要加锁的，

目的是为了避免Copy 出N 个副本出来，导致并发写。

注意：CopyOnWriteArrayList 的整个add 操作都是在锁的保护下进行的。也就是说add 方法是线程安全的。

public E get(int index) {
return get(getArray(), index);
}

这样做的好处是我们可以对CopyOnWrite 容器进行并发的读，当然，这里读到的数据可能不是最新的。

因为写时复制的思想是通过延时更新的策略来实现数据的最终一致性的，并非强一致性。

所以CopyOnWrite 容器是一种读写分离的思想，读和写不同的容器

而Vector 在读写时，使用同一个容器，读写互斥，同时只能做一件事儿。

因为通常需要复制整个基础数组，所以可变操作的开销很大

迭代器，

使用迭代器进行遍历的速度很快，并且不会与其他线程发生冲突。

在构造迭代器时，迭代器依赖于不变的数组快照。

和ArrayList基本一模一样，但它是线程安全的

CopyOnWriteArrayList 相当于线程安全的ArrayList

CopyOnWriteArrayList 使用了一种叫写时复制的方法

当有新元素add 到CopyOnWriteArrayList 时，先从原有的数组中拷贝一份出来，然后在新的数组做写操作，写完之后，再将原来的数组引用指向到新数组。

任何对array在结构上有所改变的操作（add、remove、clear等），CopyOnWriterArrayList都会copy现有的数据，再在copy的数据上修改，这样就不会影响COWIterator中的数据了，不会抛ConcurrentModificationException异常(对set,add没有作用，因为set,add本来就要加锁)，修改完成之后改变原有数据的引用即可。

读操作不加锁，写操作加锁，在进行add,set等操作时，会通过ReentrantLock进行加锁

1.复制的数组会消耗内存

2.不能读取实时性的数据

3.会产生大量的对象

Future就是：对具体的Runnable或者Callable任务的执行结果进行一系列的操作 

必要时可通过get方法获取执行结果， 这个方法会阻塞直到执行结束。

Future中的主要方法有

因为Future只是一个接口， 所以是无法直接用来创建对象使用的， 因此就有了下面的FutureTask。

FutureTask实现了RunnableFuture接口

RunnableFuture接口是什么

FutureTask也可以用作闭锁， 它可以处于以下三种状态

FutureTask在Executor框架中表示异步任务，

此外还可以表示一些时间较长的计算， 这些计算可以在使用计算结果之前启动。