Go语言视频零基础入门到精通

admin

java.util.concurrent 概述
JDK5.0 以后的版本都引入了高级并发特性，大多数的特性在java.util.concurrent 包中，是专门用于多线并发编程的，充分利用了现代多处理器和多核心系统的功能以编写大规模并发应用程序。主要包含原子量、并发集合、同步器、可重入锁，并对线程池的构造提供
了强力的支持。
原子量：是定义了支持对单一变量执行原子操作的类。所有类都有get 和set 方法，工作方法和对volatile 变量的读取和写入一样。
并发集合：是原有集合框架的补充，为多线程并发程序提供了支持。主要有：BlockingQueue，ConcurrentMap，ConcurrentNavigableMap。
同步器：提供了一些帮助在线程间协调的类，包括semaphores,barriers,latches,exchangers 等。
可重入锁：一般同步代码依靠内部锁（隐式锁），这种锁易于使用，但是有很多局限性。新的Lock对象支持更加复杂的锁定语法。和隐式锁(利用关键字synchronized加锁)类似，每一时刻只有一个线程能够拥有Lock 对象，通过与其相关联的Condition 对象，Lock 对象也支持wait 和notify 机制。
线程池：线程完成的任务（Runnable 对象）和线程对象（Thread）之间紧密相连。适用于小型程序，在大型应用程序中，把线程管理和创建工作与应用程序的其余部分分离开更有意义。线程池封装线程管理和创建线程对象。

原子量

近来关于并发算法的研究主要焦点是无锁算法（nonblocking algorithms），这些无锁算法使用低层原子化的机器指令，例如使用compare-and-swap（CAS）代替锁保证并发情况下数据的完整性。无锁算法广泛应用于操作系统与JVM中，比如线程和进程的调度、垃圾收集、实现锁和其他并发数据结构。
在 JDK5.0 之前，如果不使用本机代码，就不能用 Java 语言编写无等待、无锁定的算法。在 java.util.concurrent 中添加原子变量类之后，这种情况发生了变化。本节了解这些新
类开发高度可伸缩的无阻塞算法。

要使用多处理器系统的功能，通常需要使用多线程构造应用程序。但是正如任何编写并发应用程序的人可以告诉你的那样，要获得好的硬件利用率，只是简单地在多个线程中分割工作是不够的，还必须确保线程确实大部分时间都在工作，而不是在等待更多的工作，或等待锁定共享数据结构。如果线程之间不需要协调，那么几乎没有任务可以真正地并行。以线程池为例，其中执行的任务通常相互独立。如果线程池利用公共工作队列，则从工作队列中删除元素或向工作队列添加元素的过程必须是线程安全的，并且这意味着要协调对头、尾或节点间链接指针所进行的访问。正是这种协调导致了所有问题。

锁同步法
在 Java 语言中，协调对共享字段访问的传统方法是使用同步，确保完成对共享字段的所有访问，同时具有适当的锁定。通过同步，可以确定（假设类编写正确）具有保护一组访问变量的所有线程都将拥有对这些变量的独占访问权，并且以后其他线程获得该锁定时，将可以看到对这些变量进行的更改。弊端是如果锁定竞争太厉害（线程常常在其他线程具有锁定时要求获得该锁定），会损害吞吐量，因为竞争的同步非常昂贵。对于现代 JVM 而言，无竞争的同步现在非常便宜。
基于锁的算法的另一个问题是：如果延迟具有锁的线程（因为页面错误、计划延迟或其他意料之外的延迟），则没有要求获的锁的线程可以继续运行。还可以使用 volatile 变量来以比同步更低的成本存储共享变量，但它们有局限性。虽然可以保证其他变量可以立即看到对volatile 变量的写入，但无法呈现原子操作的读-修改-写顺序，这意味着volatile 变量无法用来可靠地实现互斥（互斥锁定）或计数器。下面以实现一个计数器为例。通常情况下一个计数器要保证计数器的增加，减少等操作需要保持原子性，使类成为线程安全的类，从而确保没有任何更新信息丢失，所有线程都看到计数器的最新值。使用内部锁实现的同步代码一般如下：

1. package jdkapidemo;
2. public class SynchronizedCounter {
3.         private int value;
4.         public synchronized int getValue() {
5.                 return value;
6.         }
7.         public synchronized int increment() {
8.                 return ++value;
9.         }
10.         public synchronized int decrement() {
11.                 return --value;
12.         }
13. }

复制代码

increment() 和 decrement() 操作是原子的读-修改-写操作，为了安全实现计数器，必须使用当前值，并为其添加一个值，或写出新值，所有这些均视为一项操作，其他线程不能打

断它。否则，如果两个线程试图同时执行增加，操作的不幸交叉将导致计数器只被实现了一次，而不是被实现两次。（注意，通过使值变量成为volatile 变量并不能可靠地完成这项操作。）计数器类可以可靠地工作，在竞争很小或没有竞争时都可以很好地执行。然而，在竞争激烈时，这将大大损害性能，因为JVM 用了更多的时间来调度线程，管理竞争和等待线程队列，而实际工作（如增加计数器）的时间却很少。使用锁，如果一个线程试图获取其他线程已经具有的锁，那么该线程将被阻塞，直到该锁可用。此方法具有一些明显的缺点，其中包括当线程被阻塞来等待锁时，它无法进行其他任何操作。如果阻塞的线程是高优先级的任务，那么该方案可能造成非常不好的结果（称为优先级倒置的危险）。

使用锁还有一些其他危险，如死锁（当以不一致的顺序获得多个锁时会发生死锁）。甚至没有这种危险，锁也仅是相对的粗粒度协调机制，同样非常适合管理简单操作，如增加计数器或更新互斥拥有者。如果有更细粒度的机制来可靠管理对单独变量的并发更新，则会更好一些；在大多数现代处理器都有这种机制。


比较并交换(CAS

大多数现代处理器都包含对多处理的支持。当然这种支持包括多处理器可以共享外部设备和主内存，同时它通常还包括对指令系统的增加来支持多处理的特殊要求。特别是，几乎每个现代处理器都有通过可以检测或阻止其他处理器的并发访问的方式来更新共享变量的指令。现在的处理器（包括 Intel 和 Sparc 处理器）使用的最通用的方法是实现名为“比较并交换（Compare And Swap）”或 CAS 的原语。（在 Intel 处理器中，比较并交换通过cmpxchg 系列指令实现。PoweRPC 处理器有一对名为“加载并保留”和“条件存储”的指
令，它们实现相同的目地；MIPS 与 PowerPC 处理器相似，除了第一个指令称为“加载链接”。）
CAS 操作包含三个操作数—— 内存位置（V）、预期原值（A）和新值(B)。如果内存位置的值与预期原值相匹配，那么处理器会自动将该位置值更新为新值。否则，处理器不做任何操作。无论哪种情况，它都会在 CAS 指令之前返回该位置的值。（在 CAS 的一些特殊情况下将仅返回 CAS 是否成功，而不提取当前值。）CAS 有效地说明了“我认为位置 V应该包含值 A；如果包含该值，则将 B 放到这个位置；否则，不要更改该位置，只告诉我这个位置现在的值即可。”
通常将 CAS 用于同步的方式是从地址 V 读取值 A，执行多步计算来获得新值 B，然后使用 CAS 将 V 的值从 A 改为 B。如果 V 处的值尚未同时更改，则 CAS 操作成功。
类似于 CAS 的指令允许算法执行读-修改-写操作，而无需害怕其他线程同时修改变量，因为如果其他线程修改变量，那么 CAS 会检测它（并失败），算法可以对该操作重新
计算。下面的程序说明了 CAS 操作的行为（而不是性能特征），但是 CAS 的价值是它可以在硬件中实现，并且是极轻量级的（在大多数处理器中）。后面我们分析Java 的源代码可以知道，JDK 在实现的时候使用了本地代码。下面的代码说明CAS 的工作原理（为了便于说明，用同步语法表示）。

1. package jdkapidemo;
2. public class SimulatedCAS {
3.         private int value;
4.         public synchronized int getValue() {
5.                 return value;
6.         }
7.         public synchronized int compareAndSwap(int expectedValue, int newValue) {
8.                 if (value == expectedValue)
9.                         value = newValue;
10.                 return value;
11.         }
12. }

复制代码

基于 CAS 的并发算法称为“无锁定算法”，因为线程不必再等待锁定（有时称为互斥或关键部分，这取决于线程平台的术语）。无论 CAS 操作成功还是失败，在任何一种情况

中，它都在可预知的时间内完成。如果 CAS 失败，调用者可以重试 CAS 操作或采取其他适合的操作。下面的代码显示了重新编写的计数器类来使用 CAS 替代锁定：

1. package jdkapidemo;
2. public class CasCounter {
3.         private SimulatedCAS value;
4.         public int getValue() {
5.                 return value.getValue();
6.         }
7.         public int increment() {
8.                 int oldValue = value.getValue();
9.                 while (value.compareAndSwap(oldValue, oldValue + 1) !=oldValue)
10.                         oldValue = value.getValue();
11.                 return oldValue + 1;
12.         }
13. }

复制代码

如果每个线程在其他线程任意延迟（或甚至失败）时都将持续进行操作，就可以说该算法是“无等待”的。“无锁定算法”要求某个线程总是执行操作。（无等待的另一种定义是保证每个线程在其有限的步骤中正确计算自己的操作，而不管其他线程的操作、计时、交叉或速度。这一限制可以是系统中线程数的函数；例如，如果有 10 个线程，每个线程都执行一次CasCounter.increment()操作，最坏的情况下，每个线程将必须重试最多九次，才能完成增加。）

无锁算法（也称为无阻塞算法）进行了大量研究，许多人通用数据结构已经发现了无阻塞算法。无阻塞算法被广泛用于操作系统和 JVM级别，进行诸如线程和进程调度等任务。虽然它们的实现比较复杂，但相对于基于锁的备选算法，它们有许多优点：可以避免优先级倒置和死锁等危险，竞争比较便宜，协调发生在更细的粒度级别，允许更高程度的并行机制等等。

原子变量类

java.util.concurrent.atomic 包中添加原子变量类。所有原子变量类都公开“比较并设置”原语语都是使用平台上可用的最快本机结构（比较并交换、加载链接/条件存储，最坏的情况下是旋转锁）来实现的。 java.util.concurrent.atomic 包中提供了原子变量的 9 种风格（AtomicInteger、AtomicLong、AtomicBoolean、AtomicReference、原子整型、长型、引用、及原子标记引用和戳记引用类的数组形式，其原子地更新一对值）。原子变量类可以认为是volatile变量的泛化，它扩展了volatile变量的概念，来支持原子条件的比较并设置更新。读取和写入原子变量与读取和写入对volatile变量的访问具有相同的存取语义。

虽然原子变量类表面看起来与SynchronizedCounter 例子一样，但相似仅是表面的。在表面之下，原子变量的操作会变为平台提供的用于并发访问的硬件原语，比如比较并交换。更多调整具有竞争的并发应用程序的可伸缩性的通用技术是降低使用的锁对象的粒度，希望的锁请求从竞争变为不竞争。从锁转换为原子变量可以获得相同的结果，通过切换为更细粒度的协调机制，竞争的操作就更少，从而提高了吞吐量。下面的程序是使用原子变量后的计数器：

1. package jdkapidemo;
2. import java.util.concurrent.atomic.AtomicInteger;
3. public class AtomicCounter {
4.         private AtomicInteger value = new AtomicInteger();
5.         public int getValue() {
6.                 return value.get();
7.         }
8.         public int increment() {
9.                 return value.incrementAndGet();
10.         }
11.         public int increment(int i) {
12.                 return value.addAndGet(i);
13.         }
14.             public int decrement() {
15.                 return value.decrementAndGet();
16.         }
17.         public int decrement(int i) {
18.                 return value.addAndGet(-i);
19.         }
20. }
21. 测试类：
22. package jdkapidemo;
23. public class AtomicCounterTest extends Thread {
24.         AtomicCounter counter;
25.         public AtomicCounterTest(AtomicCounter counter) {
26.                 this.counter = counter;
27.         }
28.         @Override
29.         public void run() {
30.                 int i = counter.increment();
31.                 System.out.println("generated number:" + i);
32.         }
33.         public static void main(String[] args) {
34.                 AtomicCounter counter = new AtomicCounter();
35.                 for (int i = 0; i < 10; i++) {//10个线程
36.                         new AtomicCounterTest(counter).start();
37.                 }
38.         }
39. }
40. 运行结果如下：
41. generated number:1
42. generated number:2
43. generated number:3
44. generated number:4
45. generated number:5
46. generated number:7
47. generated number:6
48. generated number:9
49. generated number:10
50. generated number:8

复制代码

会发现10 个线程运行中，没有重复的数字，原子量类使用本机CAS 实现了值修改的原子性。

使用原子量实现银行取款：

1. public class Account {
2.         private double balance;
4.         public Account(double money) {
5.                 balance = money;
6.                 System.out.println("Totle Money: " + balance);
7.         }
9.         public void deposit(double money) {
10.                 balance = balance + money;
11.         }
13.         public void withdraw(double money, int delay) {
14.                 if (balance >= money) {
15.                         try {
16.                                 Thread.sleep(delay);
17.                                 balance = balance - money;
18.                                 System.out.println(Thread.currentThread().getName()
19.                                                 + " withdraw " + money + " successful!" + balance);
20.                         } catch (InterruptedException e) {
21.                         }
22.                 } else
23.                         System.out.println(Thread.currentThread().getName()
24.                                         + " balance is not enough, withdraw failed!" + balance);
25.         }
26. }

复制代码

为了测试帐户类，定义一个测试类

1. public class AccountThread extends Thread {
2.         Account account;
3.         int delay;
5.         public AccountThread(Account acount, int delay) {
6.                 this.account = acount;
7.                 this.delay = delay;
8.         }
10.         public void run() {
11.                 account.withdraw(100, delay);
12.         }
14.         public static void main(String[] args) {
15.                 Account acount = new Account(100);
16.                 AccountThread acountThread1 = new AccountThread(acount, 1000);
17.                 AccountThread acountThread2 = new AccountThread(acount, 0);
18.                 acountThread1.start();
19.                 acountThread2.start();
20.         }
21. }

复制代码

运行结果如下：

Totle Money: 100.0

Thread-1 withdraw 100.0 successful!0.0

Thread-0 withdraw 100.0 successful!-100.0

从运行结果可以看出，总额100 元，使用两个线程同时取钱，都成功，最后帐户余额为-100元，表现为透支，这样破坏了数据的完整性。
从程序可以看出withdrawal 方法包含了余额判断语句，为什么还会发生数据的一致性被破坏呢？因多线程并发，当执行“balance= balance – money”这条语句时，balance的实际值已经不是先前的值。按照正确的业务逻辑，需要保证在一个取款操作结束时，不能执行另一个取款操作，需要把withdraw 同步起来，我们可使用synchronized 关键字。修改如下：
public synchronized void withdraw(double money, int delay)
运行修改后的程序,结果如下：
Totle Money: 100.0
Thread-1 withdraw 100.0 successful!0.0
Thread-0balance is not enough, withdraw failed!0.0
前面我们讲过了原子量的使用，现在修改 balance 为原子量。用原子量的特性实现取款操作的原子性。把Account 类修为AtomicAccount，把balance 定义为AtomicLong 类型，然后修改withdraw 方法，把原来方法的修改语句“balance = balance – money”修改为“balance.compareAndSet(oldvalue, oldvalue - money)”，这个方法在执行的时候是原子化的，首先比较所读取的值是否和被修改的值一致，如果一致则执行原子化修改，否则失败。如果帐余额在读取之后，被修改了，则compareAndSet 会返回FALSE，则余额修改失败，不能完成取款操作。

1. package jdkapidemo.bank;
3. import java.util.concurrent.atomic.AtomicLong;
5. public class AtomicAccount {
6.         AtomicLong balance;
8.         public AtomicAccount(long money) {
9.                 balance = new AtomicLong(money);
10.                 System.out.println("Totle Money: " + balance);
11.         }
13.         public void deposit(long money) {
14.                 balance.addAndGet(money);
15.         }
17.         public void withdraw(long money, int delay) {
18.                 long oldvalue = balance.get();
19.                 if (oldvalue >= money) {
20.                         try {
21.                                 Thread.sleep(delay);
22.                         } catch (InterruptedException e) {
23.                                 e.printStackTrace();
24.                         }
25.                         if (balance.compareAndSet(oldvalue, oldvalue - money)) {
26.                                 System.out.println(Thread.currentThread().getName()
27.                                                 + " withdraw " + money + " successful!" + balance);
28.                         } else {
29.                                 System.out.println(Thread.currentThread().getName()
30.                                                 + "thread concurrent, withdraw failed!" + balance);
31.                         }
32.                 } else {
33.                         System.out.println(Thread.currentThread().getName()
34.                                         + " balance is not enough,withdraw failed!" + balance);
35.                 }
36.         }
38.         public long get() {
39.                 return balance.get();
40.         }
41. }

复制代码

重新定义测试类：

1. package jdkapidemo.bank;
3. public class AtomicAccountTest extends Thread {
4.         AtomicAccount account;
5.         int delay;
7.         public AtomicAccountTest(AtomicAccount account, int delay) {
8.                 this.account = account;
9.                 this.delay = delay;
10.         }
12.         public void run() {
13.                 account.withdraw(100, delay);
14.         }
16.         public static void main(String[] args) {
17.                 AtomicAccount account = new AtomicAccount(100);
18.                 AtomicAccountTest accountThread1 = new AtomicAccountTest(account, 1000);
19.                 AtomicAccountTest accountThread2 = new AtomicAccountTest(account, 0);
20.                 accountThread1.start();
21.                 accountThread2.start();
22.         }
23. }

复制代码

运行结果如下：
Totle Money: 100
Thread-1 withdraw 100 successful!0
Thread-0 thread concurrent, withdraw failed!0


从运行结果可以看出，两个线程在执行 withdraw 方法时，开始余额比较都是成功的，随后在更新余额是我们使用了balance.compareAndSet(oldvalue, oldvalue - money)原子方法，
这个方法在修改余额值之前还要比较所读取的值是否和被修改的值一致，如果一致则修改，如果不一致则修改失败，返回false。并且保证在修改的过程是原子性的，不会被中断。