并发编程4：Java 中的并发基础构建模块

短命女 2024-03-23 23:31 78阅读 0赞

**目录**

1、同步容器类

1.1 - 同步容器类的问题

1.2 - 迭代和容器加锁

2、并发容器类

2.1 - ConcurrentHashMap 类

2.2 - CopyOnWriteArrayList 类

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

3.1 - 串行线程封闭

4、阻塞方法与中断方法

5、同步工具类

5.1 - 闭锁 -> CountDownLatch

5.2 - 使用 FutureTask 做闭锁

5.3 - 信号量 -> Semaphore

5.4 - 栅栏 -> Barrier

5.6 - 构建高效且可伸缩的结果缓存

--------------------

委托是创建线程安全类的一个最有效的策略：只需让现有的线程安全类管理所有的状态即可。Java 平台类库包含了丰富的并发基础构建模块，例如线程安全的容器类以及各种用于协调多个相互协作的线程控制流的同步工具类(Synchronizer)等。

Java 并发类的 API 文档，[点击这里][Link 1]。

#### 1、同步容器类 ####

同步容器类包括 *Vector* 和 *Hashtable*，二者是早期 JDK 的一部分，这些同步的封装器类是由 *Collections.synchronizedXxx* 等工厂方法创建的。这些类实现线程安全的方式是：将它们的状态封装起来，并对每个公有方法都进行同步，使得每次只有一个线程能访问容器的状态

##### 1.1 - **同步容器类的问题** #####

同步容器类都是线程安全的，但在某些情况下的*复合操作*也需要额外的客户端加锁来保护。容器上常见的复合操作包括：**迭代**(反复访问元素，直到遍历完容器中所有元素)、**跳转**(根据指定顺序找到当前元素的下一个元素)以及**条件运算**，例如 “若没有则添加” (检查在 Map 中是否存在键值 K，如果没有，就加入二元组(K，V))。//目前的很多并发容器也没有完全解决复合操作的问题，此外同步容器的加锁粒度比较大

##### 1.2 - 迭代和容器加锁 #####

在迭代期间对容器进行加锁。如果容器的规模很大，或者在每个元素上执行操作的时间很长，那么这些线程将长时间等待。即使不存在饥饿或者死锁等风险，长时间地对容器加锁也会降低程序的可伸缩性。持有锁的时间越长，那么在锁上的竞争就可能越激烈，如果许多线程都在等待锁被释放，那么将极大地降低吞吐量和 CPU 的利用率。//对整个容器加锁的弊端

如果不希望在迭代期间对容器加锁，那么一种替代方法就是“克隆”容器，并在副本上进行迭代。由于副本被封闭在线程内，因此其他线程不会在选代期间对其进行修改，这样就*避免抛出 ConcurrentModificationException* (在克隆过程中仍然需要对容器加锁)。在克隆容器时存在显著的性能开销。这种方式的好坏取决于多个因素，包括容器的大小，在每个元素上执行的工作，迭代操作相对于容器其他操作的调用频率，以及在响应时间和吞吐量等方面的需求。//创建副本/快照对容器进行迭代是一种不错的思路

#### 2、并发容器类 ####

Java 5.0 提供了多种并发容器类来改进同步容器的性能。同步容器将所有对容器状态的访向都串行化，以实现它们的线程安全性。这种方法的代价是严重降低并发性，当多个线程竞争  
容器的锁时，吞叶量将严重减低。//同步容器类性能低

![2e42d878fe27442a88bd079c5292895d.png][]

另一方面，并发容器是针对多个线程并发访问设计的。在 Java 5.0 中增加了 *ConcurrentHashMap*，用来替代同步且基于散列的 Map，以及 *CopyOnWriteArrayList*，用于在遍历操作为主要操作的情况下代替同步的 List。在新的 ConcurrentMap 接口中增加了对一些常见复合操作的支持，例如“若没有则添加”、替换以及有条件删除等。//学习并发容器类重要的是操作和使用

> 通过并发容器来代替同步容器，可以极大地提高伸缩性并降低风险。

##### 2.1 - ConcurrentHashMap 类 #####

与 HashMap 一样，ConcurrentHashMap 也是一个基于散列的 Map，但它使用了一种完全不同的加锁策略来提供更高的并发性和伸缩性。ConcurrentHashMap 并不是将每个方法都在同一个锁上同步并使得每次只能有一个线程访向容器，而是使用一种粒度更细的加锁机制来实现更大程度的共享，这种机制称为分段锁 (Lock Striping)。在这种机制中，任意数量的读取线程可以并发地访问 Map，执行读取操作的线程和执行写入操作的线程可以并发地访问 Map，并且一定数量的写入线程可以并发地修改 Map。ConcurrentHashMap 带来的结果是，在并发访问环境下将实现更高的吞吐量，而在单线程环境中只损失非常小的性能。//与HashTable相比具有更好的并发性能

ConcurrentHashMap 与其他并发容器一起增强了同步容器类：它们提供的选代器不会抛出ConcurrentModificationException，因此不需要在选代过程中对容器加锁。ConcurrentHashMap 返回的选代器具有弱一致性 (Weakly Consistent)，而并非“及时失败”。弱一致性的选代器可以容忍并发的修改，当创建迭代器时会遍历已有的元素，并可以(但是不保证)在选代器被构  
造后将修改操作反映给容器。//只能反映暂时状态

##### 2.2 - CopyOnWriteArrayList 类 #####

CopyOnWriteArrayList 用于替代同步 List，在某些情况下它提供了更好的并发性能，并且在迭代期间不需要对容器进行加锁或复制。(类似地，CopyOnWriteArraySet 的作用是替代同步 Set)

*“写入时复制(Copy-0n-Write)”*容器的线程安全性在于，只要正确地发布一个事实不可变的对象，那么在访问该对象时就不再需要进一步的同步。在每次修改时，都会创建并重新发布一个新的容器副本，从而实现可变性。“写入时复制”容器的迭代器保留一个指向底层基础数组的引用，这个数组当前位于迭代器的起始位置，由于它不会被修改，因此在对其进行同步时只需确保数组内容的可见性。因此，多个线程可以同时对这个容器进行迭代，而不会彼此干扰或者与修改容器的线程相互干扰。“写人时复制”容器返回的迭代器不会抛出ConcurrentModificationException，并且返回的元素与迭代器创建时的元素完全一致，而不必考虑之后修改操作所带来的影响。

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

***阻塞队列***支持*生产者 - 消费者*这种设计模式。该模式将“找出需要完成的工作”与“执行工作”这两个过程分离开来，并把工作项放人一个“待完成”列表中以便在随后处理，而不是找出后立即处理。生产者 - 消费者模式能简化开发过程，因为它消除了生产者类和消费者类之间的代码依赖性，此外，该模式还将生产数据的过程与使用数据的过程解耦开来以简化工作负载的管理，因为这两个过程在处理数据的速率上有所不同。//BlockingQueue

> 在构建高可靠的应用程序时，有界队列是一种强大的资源管理工具：它们能抑制并防止产生过多的工作项，使应用程序在负荷过载的情况下变得更加健壮。

Java 中的阻塞队列：

![53bedca4610b44ad9ef470e6915188da.png][]

##### 3.1 - 串行线程封闭 #####

在 java.utilconcurrent 中实现的各种阻塞队列都包含了足够的内部同步机制，从而安全地将对象从生产者线程发布到消费者线程。

对于可变对象，生产者 -消费者这种设计与阻塞队列一起，促进了*串行线程封闭*，从而将对象所有权从生产者交付给消费者。线程封闭对象只能由单个线程拥有，但可以通过安全地发布该对象来“转移”所有权。在转移所有权后，也只有另一个线程能获得这个对象的访问权限，并且发布对象的线程不会再访问它。这种安全的发布确保了对象状态对于新的所有者来说是可见的，并且由于最初的所有者不会再访问它，因此对象将被封闭在新的线程中。新的所有者线程可以对该对象做任意修改，因为它具有独占的访问权。

**对象池**利用了串行线程封闭，将对象“借给”一个请求线程。只要对象池包含足够的内部同步来安全地发布池中的对象，并且只要客户代码本身不会发布池中的对象，或者在将对象返回给对象池后就不再使用它，那么就可以安全地在线程之间传递所有权。//线程封闭就是在封闭对象上始终只有一个线程进行操作（串行化）

#### 4、阻塞方法与中断方法 ####

Thread 提供了 interrupt 方法，用于中断线程或者查询线程是否已经被中断。每个线程都有一个布尔类型的属性，表示线程的中断状态，当中断线程时将设置这个状态。

中断是一种协作机制。一个线程不能强制其他线程停止正在执行的操作而去执行其他的操作。当线程 A 中断 B 时，A 仅仅是要求 B 在执行到某个可以暂停的地方停止正在执行的操作一一前提是如果线程 B 愿意停止下来。虽然在 API 或者语言规范中并没有为中断定义任何特定应用级别的语义，但最常使用中断的情况就是取消某个操作。方法对中断请求的响应度越高，就越容易及时取消那些执行时间很长的操作。

当在代码中调用了一个将抛出 InterruptedException 异常的方法时，你自己的方法也就变成了一个阻塞方法，并且必须要处理对中断的响应。对于库代码来说，有两种基本选择：//处理中断的策略

1.  **传递 InterruptedException**。避开这个异常通常是最明智的策略，只需把 InterruptedException 传递给方法的调用者。传递 IterruptedException 的方法包括，根本不捕获该异常，或者捕获该异常，然后在执行某种简单的清理工作后再次抛出这个异常。
2.  **恢复中断**。有时候不能抛出InterruptedException，例如当代码是 Runnable 的一部分时，在这些情况下，必须捕获 InterruptedException，并通过调用当前线程上的 interrupt 方法恢复中断状态，这样在调用栈中更高层的代码将看到引发了一个中断。

#### 5、同步工具类 ####

所有的同步工具类都包含一些特定的结构化属性：它们封装了一些状态，这些状态将决定执行同步工具类的线程是继续执行还是等待，此外还提供了一些方法对状态进行操作，以及另一些方法用于高效地等待同步工具类进入到预期状态。

Java 中提供的一些同步工具类：

![9feee7faa06a4d2f93c1845d3383e4ae.png][]

##### 5.1 - 闭锁 -> CountDownLatch #####

闭锁是一种同步工具类，可以延迟线程的进度直到其到达终止状态。闭锁的作用相当于一扇门：在闭锁到达结束状态之前，这扇门一直是关闭的，并且没有任何线程能通过当到达结束状态时，这扇门会打开并允许所有的线程通过。当闭锁到达结束状态后，将不会再改变状态，因此这扇门将永远保持打开状态。闭锁可以用来确保某些活动直到其他活动都完成后才继续执行。//闭锁一旦打开就不再关闭，-> CountDownLatch

CountDownLatch 是一种灵活的闭锁实现，可以在上述各种情况中使用，它可以使一个或多个线程等待一组事件发生。闭锁状态包括一个计数器，该计数器被初始化为一个正数，表示需要等待的事件数量。countDown 方法递减计数器，表示有一个事件已经发生了，而 await 方法等待计数器达到零，这表示所有需要等待的事件都已经发生。如果计数器的值非零，那么 await 会一直阻塞直到计数器为零，或者等待中的线程中断，或者等待超时。//闭锁的实现原理，内置事件计数器

##### 5.2 - 使用 FutureTask 做闭锁 #####

FutureTask 也可以用做闭锁。(FutureTask 实现了 Future 语义，表示一种抽象的可生成结果的计算)。FutureTask 表示的计算是通过 Callable 来实现的，相当于一种可生成结果的 Runnable，并且可以处于以下 3 种状态：等待运行 (Waiting to run)，正在运行(Running)和运行完成(Completed)。“执行完成”表示计算的所有可能结束方式，包括正常结束、由于取消而结束和由于异常而结束等。当FutureTask 进入完成状态后，它会永远停止在这个状态上。//FutureTask表示可生成结果的计算

Future.get 的行为取决于任务的状态。如果任务已经完成，那么 get 会立即返回结果，否则get 将阻塞直到任务进入完成状态，然后返回结果或者抛出异常。FutureTask 将计算结果从执行计算的线程传递到获取这个结果的线程，而 FutureTask 的规范确保了这种传递过程能实现结果的安全发布。//将计算和获取结果进行分离，获取结果的过程优点类似于从阻塞队列中获取数据

下列代码中，Preloader 就使用了 FutureTask 来执行一个高开销的计算，并且计算结果将在稍后使用。通过提前启动计算，可以减少在等待结果时需要的时间。

public class Preloader {
    
        ProductInfo loadProductInfo() throws DataLoadException {
            return null;
        }
    
        private final FutureTask<ProductInfo> future = new FutureTask<>(() -> loadProductInfo());
    
        private final Thread  thread = new Thread(future);
    
        public void start() {
            //执行任务
            thread.start();
        }
    
        public ProductInfo get() throws DataLoadException, InterruptedException {
            try {
                //获取结果
                return future.get();
            } catch (ExecutionException e) {
                Throwable cause = e.getCause();
                if (cause instanceof DataLoadException) {
                    throw (DataLoadException) cause;
                } else {
                    throw LaunderThrowable.launderThrowable(cause);
                }
            }
        }
    
        interface ProductInfo {
        }
    }
    
    class DataLoadException extends Exception {
    
    }

##### 5.3 - 信号量 -> Semaphore #####

计数信号量(Counting Semaphore) 用来控制同时访问某个特定资源的操作数量，或者同时执行某个指定操作的数量。计数信号量还可以用来实现某种资源池，或者对容器施加边界。

Semaphore 中管理着一组虚拟的许可 (permit)，许可的初始数量可通过构造函数来指定。在执行操作时可以首先获得许可(只要还有剩余的许可)，并在使用以后释放许可。如果没有许可，那么 acquire 将阻塞直到有许可(或者直到被中断或者操作超时)。release 方法将返回一个许可给信号量。计算信号量的一种简化形式是二值信号量，即初始值为 1 的 Semaphore。二值信号量可以用做互斥体(mutex)，并具备不可重人的加锁语义：谁拥有这个唯一的许可，谁就拥有了互斥锁。//信号量不可重入，所以信号量不能用来替换可重入锁

##### 5.4 - 栅栏 -> Barrier #####

我们已经看到通过闭锁来启动一组相关的操作，或者等待一组相关的操作结束。闭锁是一次性对象，一旦进入终止状态，就不能被重置。

***栅栏(Barrier)***类似于闭锁，它能阻塞一组线程直到某个事件发生。栅栏与闭锁的关键区别在于，所有线程必须同时到达栅栏位置，才能继续执行。闭锁用于等待事件，而栅栏用于等待其他线程。//栅栏等待线程，闭锁等待事件（事件具有一次性）

***CyclicBarrier*** 可以使一定数量的参与方反复地在栅栏位置汇集，它在并行选代算法中非常有用：这种算法通常将一个问题拆分成一系列相互独立的子问题。当线程到达栅栏位置时将调用 await 方法，这个方法将阻塞直到所有线程都到达栅栏位置。如果所有线程都到达了栅栏位置，那么栅栏将打开，此时所有线程都被释放，而栅栏将被重置以便下次使用。//栅栏可重用

另一种形式的栅栏是 ***Exchanger***，它是一种两方 (Two-Party）栅栏，各方在栅栏位置上交换数据。当两方执行不对称的操作时，Exchanger 会非常有用，例如当一个线程向缓冲区写入数据，而另一个线程从缓冲区中读取数据。这些线程可以使用 Exchanger 来汇合，并将满的缓冲区与空的缓冲区交换。当两个线程通过 Exchanger 交换对象时，这种交换就把这两个对象安全地发布给另一方。//用来处理数据交换

##### 5.6 - 构建高效且可伸缩的结果缓存 #####

几乎所有的服务器应用程序都会使用某种形式的缓存。重用之前的计算结果能降低延迟提高吞吐量，但却需要消耗更多的内存。//用空间换时间

像许多“重复发明的轮子”一样，缓存看上去都非常简单。然而，简单的缓存可能会将性能瓶颈转变成可伸缩性瓶颈，即使缓存是用于提升单线程的性能。

下边是一个高效且可伸缩的缓存的开发示例代码，用于改进一个高计算开销的函数。该示例具有很强的参考价值：

public class Memoizer <A, V> implements Computable<A, V> {
        //使用ConcurrentHashMap缓存计算结果，提升性能
        //使用ConcurrentMap<A, Future<V>>而不是ConcurrentMap<A, V>保存计算结果，避免重复计算
        private final ConcurrentMap<A, Future<V>> cache = new ConcurrentHashMap<>();
        private final Computable<A, V> c;
    
        public Memoizer(Computable<A, V> c) {
            this.c = c;
        }
    
        public V compute(final A arg) throws InterruptedException {
            while (true) {
                //1-计算时先获取，如果已经有线程计算，那么就阻塞等待获取前一个线程的计算结果
                Future<V> f = cache.get(arg);
                if (f == null) {
                    //2-封装计算任务，若没有则添加
                    Callable<V> eval = () -> c.compute(arg);
                    FutureTask<V> ft = new FutureTask<V>(eval);
                    f = cache.putIfAbsent(arg, ft); //原子性->若没有则添加
                    if (f == null) {
                        f = ft;
                        ft.run(); //3-执行任务
                    }
                }
                try {
                    //4-获取计算结果：可能会抛出异常，需对异常进行处理
                    return f.get();
                } catch (CancellationException e) {
                    //5-如果执行被取消，需要移除计算任务，给其他线程计算机会
                    cache.remove(arg, f);
                } catch (ExecutionException e) {
                    //6-执行异常直接抛出
                    throw LaunderThrowable.launderThrowable(e.getCause());
                }
            }
        }
    }
    
    interface Computable <A, V> {
        V compute(A arg) throws InterruptedException;
    }
    
    class ExpensiveFunction implements Computable<String, BigInteger> {
        public BigInteger compute(String arg) {
            return new BigInteger(arg);
        }
    }

注意：Memoizer 没有解决缓存逾期的问题，但它可以通过使用 FutureTask 的子类来解决，在子类中为每个结果指定个逾期时间，并定期扫描缓存中逾期的元素。同样，它也没有解决缓存清理的问题，即移除旧的计算结果以便为新的计算结果腾出空间，从而使缓存不会消耗过多的内存。//使用缓存，应该考虑缓存的失效问题(存储时间)

至此，全文结束。

[Link 1]: https://docs.oracle.com/en/java/javase/19/docs/api/java.base/java/util/concurrent/package-summary.html
[2e42d878fe27442a88bd079c5292895d.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/cfb890ae22684c1aaa1fecf02c65f12e.png
[53bedca4610b44ad9ef470e6915188da.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/f5df196558834be1afd04274bfb09e12.png
[9feee7faa06a4d2f93c1845d3383e4ae.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/13/b6672453b5664dc19c50870b7fb93750.png