[关闭]
@kiraSally 2018-03-12T18:59:31.000000Z 字数 8759 阅读 4452

并发番@Concurrent包一文通（1.8版）

JAVA 并发 源码 1.8版

  • 笔者个人博客 kiraSally的掘金个人博客 感谢支持
  • 致谢 <<操作系统精髓与设计原理（原书第六版）>>
  • 推荐阅读Doug Lea大师的并发包专题论文 The java.util.concurrent Synchronizer Framework

1.并发机制

1.1 并发的原理

  • 单核系统：线程交替执行，由于交替又快又多，给人一种同时执行的感觉
  • 多核系统：不仅可以交替执行线程，而且可以重叠执行线程
  • 补充： 本章指的并发主要指的是线程间的并发

1.2 常见的并发机制

常用并发机制.png-73.9kB

1.3 不同系统的并发机制

  • UNIX：管道、消息、共享内存、信号量、信号
  • Linux内核：原子操作、自旋锁、信号量、屏障（由于服务器一般都位于Linux服务器上，因此此是我们最重要要了解的）
  • Solaris线程同步原语：互斥锁、信号量、多读者/单写者锁、条件变量
  • Windows：等待函数、分派器对象、临界区、轻量级读写锁和条件变量

2.互斥

2.1 互斥的要求

  • 强制互斥： 当临界区共享时，一次只允许一个线程进入临界区，即必须强制实施互斥
  • 禁止干涉： 一个在非临界区停止的线程不能干涉其他线程，包括临界区和非临界区的线程
  • 禁止无限延迟： 决不允许出现需要访问临界区的线程被无限延迟的情况，如产生死锁或饥饿
  • 可用立入： 当没有线程在临界区中时，任何需要进入临界区的线程必须能够立即进入
  • 核数无关： 对相关线程的执行速度和处理器的数目没有任何要求和限制
  • 有限时间： 一个线程驻留在临界区的时间必须是有限的

2.2 互斥方案

  • 硬件支持：处理器原生支持的互斥指令，好处是可以减少开销，但很难成为一个通用的解决方案
  • 系统或语言级别支持：即由操作系统或程序语言提供该级别的互斥支持，比如信号量、管程、消息传递等
  • 软件方法支持：这些方法通常基于在访问内存时基本互斥的假设，尽管允许访问的顺序事先没有具体安排，但同时访问内存中的同一地址的操作被内存仲裁器串行化执行了，即可以理解用算法的方式解决互斥问题，比如Dekker算法、Peterson算法

2.3 互斥的软件方法支持

  • 综述： 软件方法是通过算法方式实现互斥，即从算法角度将临界区访问串行化，其并没有考虑硬件、操作系统或是编程语言的支持，最知名的就是Dekker算法和其的简明版本Peterson算法（算法名即是发明者）

2.3.1 Dekker算法

  1. /**
  2.  * Dekker算法基本约束：
  3.  *      某一时刻对某一内存地只能进行一次访问
  4.  * 1.设置flag做为两个线程进入临界区的密钥，当一个线程失败，其他仍可访问
  5.  *      - 每个线程只能改变自身的flag，只能检查其他线程的flag而不能改变
  6.  *      - 当一个线程要进入临界区，需周期性检查另一个线程flag，直到另一线程不在临界区
  7.  *      - 当线程进入临界区时，应立即设置自身flag为true，表明占领临界区
  8.  *      - 当线程离开临界区时，应立即设置自身flag为false，表明释放临界区
  9.  * 2.设置turn用于安排临界区的访问顺序，访问线程必须重复读取turn值直到被允许进入临界区
  10.  *      - 当turn值等于线程号，该线程可以进入其临界区
  11.  *      - 否则，该线程必须被强制等待（忙等待或自旋等待）
  12.  */
  13. public class Dekker {
  14.     //观察两个线程的状态
  15.     boolean[] flag = {false,false};
  16.     //表示临界区访问权限的轮转，初始权利给P1 -- 安排执行顺序避免谦让造成的活锁问题
  17.     int turn = 1;
  19.     public void P0(){
  20.         while (true){
  21.             //设置P0的flag为true，同时检查P1的flag
  22.             flag[0] = true;
  23.             while (flag[1]){
  24.                 //当临界区不可用时，判断当前临界区权限是否是P1
  25.                 if (turn == 1){ // 用于处理活锁问题
  26.                     //当临界区权限是P1时，需要将P0设置为false，使得P1能进入临界区，用于处理死锁问题
  27.                     flag[0] = false;
  28.                     //循环校验turn的权限(空自旋)，直到P1执行完毕将权限交给P0
  29.                     while (turn == 1){
  30.                         /** do Nothing 空自旋 **/
  31.                     }
  32.                     flag[0] = true;//此时P1应已执行完毕，应当禁止P1进入临界区
  33.                 }
  34.             }
  35.             //当P1的flag为false时，P0可以立即进入临界区
  36.             /** critical section 临界区 **/
  37.             //当临界区执行完之后，turn设置为1，将临界区访问权限交换给P1
  38.             //并将P0的flag设置为false，释放临界区，使得P1可进入临界区
  39.             turn = 1;
  40.             flag[0] = false;
  41.             /** do otherThings **/
  42.         }
  43.     }
  45.     public void P1(){
  46.         while (true){
  47.             //设置P1的flag为true，同时检查P0的flag
  48.             flag[1] = true;
  49.             while (flag[0]){
  50.                 //当临界区不可用时，判断当前临界区权限是否是P0
  51.                 if (turn == 0){ //用于处理活锁问题
  52.                     //当临界区权限是P0时，需要将P1设置为false，使得P0能进入临界区，用于处理死锁问题
  53.                     flag[1] = false;
  54.                     //循环校验turn的权限(空自旋)，直到P0执行完毕将权限交给P1
  55.                     while (turn == 0){
  56.                         /** do Nothing 空自旋 **/
  57.                     }
  58.                     flag[1] = true;//此时P0应已执行完毕，应当禁止P0进入临界区
  59.                 }
  60.             }
  61.             //当P0的flag为false时，P1可以立即进入临界区
  62.             /** critical section 临界区 **/
  63.             //当临界区执行完之后，turn设置为0，将临界区访问权限交换给P0
  64.             //同时将P1的flag设置为false，释放临界区，使得P0可进入临界区
  65.             turn = 0;
  66.             flag[1] = false;
  67.             /** do otherThings **/
  68.         }
  69.     }
  70.     public static void main(){
  71.         /** 并发执行P0 P1 读者有兴趣可自己验证一下**/
  72.     }
  73. }

2.3.2 Peterson算法

  1. /**
  2.  * Peterson算法比Dekker算法更加简单出色而且很容易推广到多个线程
  3.  *     1.互斥保护验证：P0角度
  4.  *      - 当PO设置flag[0]=true，则P1不能进入临界区
  5.  *      - 当P1已进入临界区，而flag[1]=true，P0不能进入临界区
  6.  *     2.避免相互阻塞验证：P0角度
  7.  *      - 当P0在while循环中被阻塞，此时flag[1]=true且turn=1
  8.  *      - 当flag[1]=false或turn=0，此时P0可以进入临界区
  9.  *     3.复杂度：该算法用简单的交替进入临界区的方式降低了并发互斥的复杂度
  10.  */
  11. public class Peterson {
  12.     boolean[] flag = {false,false};//表明每个互斥线程的位置
  13.     int turn = 0;//解决同时发生的冲突
  15.     public void P0(){
  16.         while (true){
  17.             flag[0] = true;
  18.             //每次都要显式设置turn=1并作为while空自旋条件，迫使其他线程也有进入临界区的机会
  19.             //这也是解决互斥的一个简洁方案，大家依次来，不能重复独占
  20.             turn = 1;
  21.             while (flag[1] && turn == 1){
  22.                 /** do Nothing 空自旋**/
  23.             }
  24.             /** critical section 临界区 **/
  25.             flag [0] = false;
  26.             /** do otherThings **/
  27.         }
  28.     }
  30.     public void P1(){
  31.         while (true){
  32.             flag[1] = true;
  33.             turn = 0;
  34.             while (flag[0] && turn == 0){
  35.                 /** do Nothing 空自旋**/
  36.             }
  37.             /** critical section 临界区 **/
  38.             flag [1] = false;
  39.             /** do otherThings **/
  40.         }
  41.     }
  42.     public static void main(){
  43.         /** 并发执行P0 P1 读者有兴趣可自己验证一下**/
  44.     }
  45. }

2.4 互斥的硬件支持

2.4.1 中断禁用

2.4.1.1 中断禁用综述
  • 原理：单核系统，并发线程不能重叠只能交替；一个线程将一直运行直到它调用一个系统服务或被中断
  • 实现：为了保证互斥，只要保证一个线程不被中断即可，可以通过系统内核为启用(enable)或禁用(disable)中断定义的原语提供
  1. //由于临界区不能被中断，因此可以保证互斥
  2. while(true){
  3.     /** disable interrupt 禁用中断 **/
  4.     /** critical zone 临界区 **/
  5.     /** enable interrupt 启用中断 **/
  6.     /** do other 其他部分 **/
  7. }
2.4.1.2 中断禁用问题
  • 效率低： 该方法的代价高，处理器被限制只能交替执行，效率明显降低
  • 不支持多核： 该方法不能用于多处理器结构，此时很可能会出现多个线程同时执行

2.4.2 专用机器指令

2.4.2.1 专用机器指令综述
  • 前提：多核系统中，处理器间共享内存，之间没有支持互斥的中断机制
  • 原理：在硬件级别上，对存储单元的访问排斥对相同单元的其他访问，在专用指令执行过程中，任何其他指令访问内存将被阻止，同时这些动作在一个指令周期内完成
  • 实现：常用的两种机器指令有exchange指令和比较和交换指令CAS(参见笔者的并发番@CAS一文通)
2.4.2.2 专用机器指令问题
  • 使用忙等待：等待时继续消耗处理器时间
  • 可能饥饿：进入临界区的线程是随机的，可能会导致某些进程被无限期拒绝进入
  • 可能死锁：比如当前线程进入临界区后被中断，而进入临界区的新线程（优先级更高）试图使用同一资源，会被拒绝从而进入忙循环等待，但由于原线程优先级低，将永远不会被调度执行

2.5 互斥的系统或语言级别支持

2.5.1 信号量

2.5.1.1 信号量概述
  • 基本原理： N个线程可以通过简单的信号进行合作，让一个线程可以被迫在某个位置停止，直到它接收到一个特殊的信号。任何复杂的合作需求都可以通过适当的信号结构得到满足
  • 组成部分：
    • 为了发信号，需要使用一个称作信号量的特殊变量sem，通常被初始化为非负数
    • 为通过信号量sem传送信号，线程可执行原语semSignal(sem)：此时信号量sem+1，当sem小于或等于0，则被semWait阻塞的线程被阻塞
    • 为了通过信号量sem接收信号，线程可执行原语semWait(sem)：此时信号量sem-1，当sem变成负数，则执行semWait的线程被阻塞，否则线程继续执行
  • 分类：无论是计数信号量还是二元信号量，都需要使用队列保存在信号量上等待的进程/线程，这就需要决定进程按照什么顺序从队列中移除
    • 强信号量：使用FIFO先见先出公平策略(即被阻塞时间最长的进程/线程最先被队列释放)的信号量(常用)
    • 弱信号量：没有规定进程/线程从队列中移除顺序的信号量
  • 补充： 二元信号量的区别只是sem的值只能是0和1而已
2.5.1.2 信号量的实现（CAS版）
  1. /**
  2.  * 设计原则：任何时候只能一个线程可以用wait和signal操作控制一个信号量
  3.  * 要求：semWait和semSingal操作必须作为原子原语实现
  4.  * semaphore信号量（以下都简称sem）的属性 
  5.  *      flag : 表示信号量是否可用，默认是0
  6.  *      count: 
  7.  *          当>=0时,表示可执行semWait而不被挂起的线程数
  8.  *          当<0时,表示挂起在信号量的等待队列的线程数
  9.  *      queue: 表示信号量相关联的等待队列，被阻塞的线程需放入该队列中
  10.  *  PS:这里我们选用Boolean版本的CAS
  11.  */
  12. semWait(sem){
  13.     //当发现sem.flag不为0时，就自旋等待直到为0
  14.     //补充一点：忙等待可以保证队列操作的同步，
  15.     //但由于wait和signal执行时间短，其开销还是很小的
  16.     while(!compare_and_swap(sem.flag,0,1));
  17.     sem.count--;
  18.     if(sem.count < 0){
  19.         /** 该线程进入sem.queue等待队列中并被阻塞 **/
  20.     }
  21.     sem.flag = 0;
  22. }
  23. semSignal(sem){
  24.     //当发现sem.flag不为0时，就自旋等待直到为0
  25.     while(!compare_and_swap(sem.flag,0,1));
  26.     sem.count++;
  27.     if(sem.count <= 0){
  28.         /** 从sem.queue等待队列中移出，被移出的线程进入就绪队列**/
  29.     }
  30.     sem.flag = 0;
  31. }
2.5.1.3 信号量实现互斥
  1. final int n = /** 线程数 **/
  2. int s = 1;//semaphore
  3. public void P(int i){
  4.     while(true){
  5.         semWait(s);
  6.         /** critical zone 临界区 **/
  7.         semSignal(s);
  8.         /** do other 其他部分 **/
  9.     }
  10. }

2.5.2 管程

2.5.2.1 管程概述
  • 定义：管程是由一个或多个过程、一个初始化序列和局部数据组成的软件模块
  • 特点：
    • 局部数据变量只能被管程的过程访问，任何外部过程都不能访问
    • 一个线程通过调用管程的一个过程进入管程
    • 在任何时候，只能有一个线程在管程中执行，调用管程的任何其他线程都将被阻塞，以等待管程可用
  • 同步机制： 对于管程而言，需要一种机制(条件变量)实现如下两个效果：
    • 当一个线程调用管程时，直到满足某个条件之前必须被挂起且能释放该管程，以让其他线程可入
    • 当条件满足且管程再次可用时，恢复该线程且允许其在挂起点重新进入管程继续执行后续操作
2.5.2.2 条件变量
  • 综述： 管程通过使用条件变量提供对同步的支持，其被管程包含且只能被管程访问
  • 操作： 有两个函数可以操作条件变量
    • cwait(c): 调用线程的执行在条件c上挂起，管程现在可被另一个线程使用
    • csignal(c)： 恢复执行在cwait之后因为某些条件而挂起的线程，如果有多个被线程被挂起，根据排队策略选择其中一个在挂起点恢复并继续执行
2.5.2.3 管程使用
  • 使用： 管程是个很有用的技术，我们可以用管程锁定任何对象，比如对类似链表的对象，可以用一个锁锁住整个链表，也可以每个表用一个锁，还可以为表中的每个元素用一个锁（联想一下Concurent包的那些并发数据结构，是不是感到很亲切？~）；同时管程通过条件变量提供对同步的支持，例如Condition，尤其适合生产者-消费者模式的业务场景处理

2.5.3 消息传递

2.5.3.1 消息传递概述
  • 消息定义：消息传递指的是线程间通过发送消息的方式实现相互通信
  • 消息实现：通常会提供一对原语实现 send(destination,message) 、receive(source,message)
  • 发送消息：一个线程以消息massage的形式给另一个指定的目标destination线程发送消息；
  • 接收消息：线程通过执行recieve原语接收来自源线程source的消息massage
2.5.3.2 消息结构
  • 消息类型： 指定的消息类型，接收者往往会根据该类型进行消息监听和捕获
  • 目标ID/源ID： 发送方/源的标识符
  • 消息长度： 整个消息的总长度，注意要控制长度
  • 控制信息： 额外信息，比如创建消息链表的指针、记录源和目标之间传递消息数目、顺序和序号以及优先级
  • 消息内容： 消息正文，相当于Body
  • 补充： 读者可以参间ISO中各种协议的包格式，比如HTTP和TCP的包

消息.jpg-127.1kB

2.5.3.3 消息通信情况
  • send：要么发送线程被阻塞直到该消息被目标线程接收，要么不阻塞
  • receive:
    • 若消息在接收之前已经被发送，该消息被目标线程接受并继续执行
    • 若没有正在等待的消息，则该目标线程被阻塞直到所等待的消息到达，或者该线程继续执行，放弃接收

消息通信情况.jpg-401.1kB

2.5.3.3 消息通信组合
  • 阻塞send，阻塞receive：发送者和接收者都被阻塞，直到完成信息的投递
  • 无阻塞send，无阻塞recive：不要求任何一方等待
  • 无阻塞send，阻塞receive：发送者发送完之后可以继续运行，但接收者被阻塞直到请求的消息到达，最典型的应用就是服务器，即允许其他服务尽快向其发送多条信息，接收线程必须在消息到达后才能继续工作，否则就被阻塞
2.5.3.4 消息通信寻址
  • 寻址： 指的是确定消息来源和去向的方式，即确认消息源和消息目标地址
  • 分类： 直接寻址和间接寻址
  • 直接寻址： send原语包含目标地址，receive原语分成两种处理方式，
    直接：要求接收者显示指定发送者，接收者必须事先知道源地址，且只接收该发送者消息
    隐式：当发送者不可预期时，recieve原语的source中必须保留接收操作执行后的返回地址
  • 间接寻址： 消息不是直接从发送者发送给接收者，而是两者共享一个数据结构，其由临时保存消息的队列组成，称之为信箱，发送者给信箱发送消息，接收者从信箱中取消息，这样就实现了发送者和接收者之间的解耦
2.5.3.5 消息通信实现互斥
  • 假设一组并发线程共享一个信箱box(用来存放消息)，该信箱被初始化成一个无内容的消息
  • 希望进入临界区的线程首先试图接收一条消息，若此时信箱为空，则该线程被阻塞
  • 一旦线程获取到消息，进入临界区，之后将该消息放回信箱
  • 消息函数可以看作是线程之间传递的一个令牌
  1. int n = /* 线程数 */
  2. void P(int i){
  3.     Message msg = /* 消息 */
  4.     while(true){
  5.         receive(box,msg);
  6.         /* 临界区 */
  7.         send(box,msg)
  8.     }
  9. }
  10. send(box,null);//信箱被初始化为一个无内容的消息

2.5.4 读/写优先

  • 读写问题： 线程共享一个数据区，有些线程只读，有些线程可读写，此时读线程不需要排斥其他读线程，但写线程需要排斥其他所有线程
    • 任意多的读线程可以同时读该数据区
    • 一次只有一个写线程可以写该数据区
    • 如果一个写线程正在写数据，禁止任何读线程的读操作
  • 读写策略： 为了提供并发性能，针对读多写少的情况，完全可以操作写者优先的策略，具体的实现等笔者的并发番@ReentrantReadWriteLock或COW再详述(哪个先写就写在哪里)

3.Concurrent并发包

3.1 Concurrent包整体架构

Concurrent包整体架构.jpg-313.4kB

3.2 Concurrent包整体类图

并发包.jpg-707.7kB

3.3 Concurrent包实现机制

  • 综述： 在整个并发包设计上，Doug Lea大师采用了3.1 Concurrent包整体架构的三层结构
  • 补充： 并发包所涉及的内容笔者会陆续推出对应番进行阐述，敬请期待（进度视笔者的忙碌程度而定）

3.3.1 底层-硬件指令支持

  • 综述： 并发包最底层是依赖于硬件级别的Volatile和CAS的支持
  • Volatile：借用Volatile的内存读写语义和阻止重排序保证数据可见性
  • CAS： 借用CAS的高效机器级别原子指令保证内存执行的 读-改-写 操作的原子性
  • 组合： 借用Volatile变量的读/写和CAS实现线程之间的有效通信，保证了原子性、可见性、有序性

3.3.2 中间层-基础数据结构+算法支持

  • 综述： 在数据结构和算法的设计使用上，Doug Lea大师专门设计了AQS框架作为所有并发类库的并发基础，同时引入非阻塞算法和原子变量类增强了并发特性
  • AQS框架： AQS中提供了最基本、有效的并发API， Doug Lea大师期望其作为所有并发操作的基础解决方案，并发包中的绝大部分实现都是依赖于AQS（AbstractQueuedSynchronizer），同时AQS的基础是CAS和Volatile的底层支持
  • 非阻塞数据结构: 非阻塞数据结构是非阻塞队列的设计基础，同时也是阻塞队列的参考对比的重要依据
  • 原子变量类： Doug Lea大师专门为所有的原子变量设计了专门的类库，甚至在后期还对齐做了增强，比如LongAdder、LongAccumulator等，从侧面可以反映出数值操作对于编程的重要性

3.3.3 高层-并发类库支持

  • 综述： Doug Lea大师在并发包中已经提供了丰富的并发类库极大方便了快速、安全的使用并发操作
  • Lock： Lock接口定义了一系列并发操作标准，详情参见 并发番@Lock接口一文通（1.8版）
  • 同步器： 每个并发类的同步器的实现依赖于AQS(继承)，比如ReentrantLock中的Sync；同时笔者也将 并发类 同属于同步器的范围内
  • 阻塞队列: 顾名思义，支持阻塞的队列，主要是以Queue结尾的类
  • 执行器： 所谓执行器，指的是任务的执行者，比如线程池和Fork-Join
  • 并发容器： 即支持并发的容器，主要包含COW和以Concurrent开头的类，通常并发容器是非阻塞的

并发番@Concurrent包一文通(1.8版) 由 黄志鹏kira 创作，采用 知识共享 署名-非商业性使用 4.0 国际 许可协议 进行许可。

本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名。

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注