您的位置:首页 > 其它

使用 Accessor Service 共享可变对象

梦江南 2021-01-13 20:46 10 查看 https://blog.51cto.com/1508239

Brian Goetz 在他的 "Java Concurrency In Practice" 中,第54页介绍了如何在线程间安全地共享对象。需要注意下面4点:


  1. 对象要保持线程限定(仅限线程内部更新),即只由拥有该对象的线程更新

  2. 共享对象时保持只读,只做一次性发布

  3. 对象内部是线程安全的,对象内部实现同步

  4. 对象由锁机制保护


本文介绍了方案4的一个变种,共享对象既不保持线程限定、也不只读或者在对象内部实现同步,而是采用读写锁确保对象状态正确。下面展示的代码可高度并发,无需使用 `synchronized` 且不会发生线程竞争造成应用性能下降。虽然没有使用 `synchronized`,通过应用特定规则仍然可以确保共享对象的修改在所有线程中可见。


1. 创建共享对象


共享的对象应遵守一些规则。这样不但能避免发生线程可见性问题,还能确保线程安全。示例如下:


```java
/**
* 安全共享对象实例
*/
public final class SharedObject {
   /**
    * 互斥读写锁
    */
   private final ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
   
   /**
    * 可变状态字段示例
    */
   private volatile String data;
   
   /**
    * 按照加锁规则声明的其他可变状态
    */
   
   /**
    * 默认包级私有构造函数
    */
   SharedObject() {
   }

   /**
    * 包级私有拷贝函数
    */
   SharedObject(SharedObject template) {
       this.data = template.data;
   }

   boolean readLock() {
       return lock.readLock().tryLock();
   }

   boolean writeLock() {
       return lock.writeLock().tryLock();
   }

   void readUnlock() {
       lock.readLock().unlock();
   }

   void writeUnlock() {
       lock.writeLock().unlock();
   }

   public String getData() {
       return data;
   }

   /**
    * 包级私有 setter 方法
    */
   void setData(String data) {
       this.data = data;
   }
}
```


对象本身包含一个 `ReentrantReadWriteLock` 用来锁定。只要没有未释放的 write-lock,就能支持多线程并发读取。我们不希望共享对象实例被不必要的锁降低读取性能。如果线程修改了对象状态,就需要确保并发读取不会因为其他线程的影响读到无效状态。当对象成功获得了 write-lock 则不允许进行读取。这就是 `ReadWriteLock` 的设计思想。


其他线程可以通过另一个 Accessor Service 服务访问 `SharedObject` 实例,接下来会讲解这个过程。但是首先,让我们站在可见性角度应用规则,把 `SharedObject` 改造成线程安全类,通过 Accessor Service 提供服务。规则如下:


  1. 只允许相同 package 的类创建对象和修改状态,这里假定类与它的 Accessor Service 位于同一包下;

  2. 不要从 Accessor Service 中丢掉原始对象,因此需要在构造函数中拷贝对象;

  3. 声明 `ReentrantReadWriteLock` 进行状态锁定;

  4. 所有修改状态的方法应只对同一个包中的类开放,比如 Accessor Service;

  5. 所有可变状态都要声明为 `volatile`;

  6. 如果 `volatile` 字段碰巧是对象引用,必须遵守以下规则:

             1.对象必须是不可变的;

             2.该字段引用的对象必须遵守规则4、5、6。


规则4、5、6能够保证应用执行过程中 `SharedObject` 对象的状态变化对所有线程可见。这样,无论对应的内存采用何种同步机制,都能够确保结论成立。遵守以上规则,可以看到对象当前最新状态,但这里并不保证状态有效。对象状态的有效性仅取决于修改操作是否线程安全,关于这部分会在接下来 Accessor Service 的使用中介绍。


2. 共享对象 Accessor Service


现在,让我们看看上面提到的 service 类,它负责为共享对象提供线程安全的更新操作:


```java
package org.projectbarbel.playground.safeaccessor;

import java.util.ConcurrentModificationException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Function;

/**
* 线程安全地访问共享对象
*/
public final class SafeAccessorService {

   /**
    * {@link ConcurrentHashMap} 对所有线程安全地发布对象,对象无一遗漏
    *
    */
   private final Map<String, SharedObject> sharedObjects = new ConcurrentHashMap<String, SharedObject>();
   
   public SafeAccessorService() {
   }

   /**
    * 线程安全地访问共享实例
    *
    * @param objectId         object id
    * @param compoundAction   共享对象上执行的原子操作
    * @param lock             lock 函数, 为对象加锁
    * @param unlock           unlock 函数, 为对象解锁
    * @return 更新后的实例
    */
   private SharedObject access(String objectId, Function<SharedObject, SharedObject> compoundAction,
           Function<SharedObject, Boolean> lock, Consumer<SharedObject> unlock) {
       // 安全地创建新实例, 由 ConcurrentHashmap 一次性发布
       SharedObject sharedObject = sharedObjects.computeIfAbsent(objectId, this::createSharedInstance);
       if (lock.apply(sharedObject)) {
           try {
               // 线程安全地修改对象, sharedObject 需要遵守加锁规则
               return compoundAction.apply(sharedObject);
           } finally {
               unlock.accept(sharedObject);
           }
       } else {
           // 交由客户端处理
           throw new ConcurrentModificationException(
                   "the shared object with id=" + objectId + " is locked - try again later");
       }
   }

   /**
    * Accessor Service 更新操作示例,可自行定义
    * @param objectId 待更新对象
    * @param data 设置给对象的数据
    */
   public void updateData(String objectId, String data) {
       access(objectId, so -> {
           so.setData(data);
           return so;
       }, so -> so.writeLock(), so -> so.writeUnlock());
   }

   /**
    * Get access 方法返回共享对象快照, 不修改原始对象,
    * 确保客户端始终工作在有效状态。
    * 传入对象 id 无效时,方法会创建一个新实例。
    *
    * @param objectId {@link SharedObject} id
    * @return 共享对象拷贝
    */
   public SharedObject get(String objectId) {
       return access(objectId, so -> new SharedObject(so), so -> so.readLock(), so -> so.readUnlock());
   }

   /**
    * 从 map 中移除对象
    *
    * @param objectId 待移除的对象 id
    */
   public void remove(String objectId) {
       sharedObjects.remove(objectId);
   }

   /**
    * 创建新的共享实例
    *
    * @param id 共享对象 id
    * @return 新创建的对象实例
    */
   private SharedObject createSharedInstance(String id) {
       return new SharedObject();
   }
}
```


共享对象存储在 `ConcurrentHashmap` 中(18行)。虽然能够保证线程创建对象时实现一次性安全发布,但不能承诺修改可变对象对所有线程可见。要实现可见性,必须在遵守上述规则的前提下,对修改状态操作使用同步技术。


`access()` 是 `SaveAccessorService` 类的核心方法,能够根据需要安全地创建新实例(36行);根据传给 `access()` 的锁定和解锁函数,有 read-lock 或 write-lock 两种类型;如果加锁成功,会在对象上调用 `compoundAction`(40行);程序的最后会释放锁(46行)。这种激进的非等待策略会被修改传入的锁定和解锁函数或者 `SharedObject` 中定义加锁方法削弱。


让我们来看 `updateData()` 方法(56行),它调用了上面提到的 `access()` 方法执行更新操作。`update` 方法再调用 `access()`,这里 `compoundAction` 会在共享对象上调用 `setData()` 方法。整个调用过程在 write-lock(60行)控制下进行。`update()` 函数只对 `data` 变量进行了一个非常简单的更新。使用者可根据需要为共享对象定义更复杂的操作。所有操作都是“自动”执行,也就是说只要向 `access()` 方法传递 `compoundAction`,就会在 write-lock 的控制下执行。


`get()`(71行)也调用了 `access()` 方法,但这里只是在 read-lock 的控制下创建了对象快照。这样能确保快照中的值一直有效,因为只要 write-lock 未释放,获取共享对象的 read-lock 就会失败。注意:`get()` 方法不会把原始对象的引用返回给客户端。这种技术有时被称作实例约束,确保不会在 Accessor Service 以外的地方对原始实例执行非线程安全的操作。


3. 优点与不足


这种模式存在优点与不足。每个对象都需要根据读写类型分配对应的锁,模式的优点在于能够把锁的控制范围减到最小。上面的示例中,加锁失败后不会等待,直接向客户端返回 `ConcurrentModificationException` 异常,交由客户端处理。客户端捕获异常后可继续处理其他任务,完成后返回。


客户端也可以不选择这种激进的加锁策略,比如让线程等待直到加锁成功。`ReentrantReadWriteLock` 提供了 `tryLock(long timeout, TimeUnit unit)` 方法可以做到这一点。可以通过 `access()` 调用时传入的 `lock` 和 `unlock` 函数进行加锁,也可以修改 `SharedObject` 中的 `lock` 函数。使用者可以决定使用其他类型的锁或者采取不同的加锁策略。因此,完全可以根据自己的加锁需求进行调整。我提出了“可扩展性选项”,出现线程争用的情况极低。


模式的另一个优点,客户端可以定义类似 `required` 的原子操作。`updateData()` 方法只是更新操作一个简单的示例,实际会用到更加复杂的操作,只需向 `SaveAccessorService` 添加类似 `updateData()` 的方法即可。与基于 Spring 的应用类似,底层是数据库可以为 service 添加多个 update 方法;同样的,这些方法会执行其他 compound action,比如在共享对象上执行多个 setter 方法。


模式还有一个优点:对象本身无需关心加锁过程,只有 getter、setter 和排他锁对象。Service 中定义了 compound action,这些 action 受对象锁保护。这种方法可以方便地添加复杂 action,甚至可以为多种不同的共享对象定义 action。Compound action 不限于某个共享对象状态,还可以是针对多个对象的原子操作。这种情况下可能引入新的多线程问题,比如死锁。


模式的缺点在于,使用者要能处理好共享对象组合。共享对象必须遵守设定的规则,否则可能出现 Accessor Service 暴露不必要的引用,结果线程读到过期数据。在我看来,另一个缺点是对象存储在 map 中,对客户端透明。如果客户端无法很好地管理 map,可能会带来内存泄漏。例如,只添加对象不移出对象,结果会造成老年代内存使用增加。可以通过 `remove()` 方法移除对象,或者调用其他方法清空整个 map。


4. 性能


这个方案是否比 `synchronized` 性能更好?要回答这个问题,首先需要知道读操作的比例是否大大超过写操作。虽然与具体的应用紧密相关,但是可以确认读写锁针对并发性能进行了设计优化。Brian Goetz 在书中提到:“多处理器系统中,访问以读为主的数据结构读写锁会进行优化;而其他场合下,由于自身实现的复杂性,性能上会比排它锁略差“(Java Concurrency In Practice, 2006, 286页)。因此,武断地评价这种方案比其他实现更好是不合适的。得出正式的结论前,需要对你的应用进行性能分析。也可以在每次更新前获得共享对象监视器,执行读操作,对比本文的加锁策略进行评估。


5. 可见性


严格来说,Accessor Service 中的 `SharedObject` 并不需要声明 `volatile`,因为显示加锁已经能够保证类似 `synchronized` 的内存可见性。然而,我们认为 `SharedObject` 可以安全地用到许多场合。例如,即使 `SafeAccessorService` 暴露了对象引用,`SharedObject` 并不会马上失败,因为它的实现遵守了前文的规则。此外,在我设计的一个 “ultra-fast” 应用中,使用 `AtomicBoolean` 作为共享对象的 psuedo-lock,这些对象由 Accessor Service 实例管理。这意味着,为了达到最高的性能和最少的线程争用,抛弃了所有 JDK 提供的复杂同步机制。这种情况下,在 `SharedObject` 上应用的规则就显得极为重要,因为我放弃了所有线程数据可见性保证,比如 `ReentrantReadWriteLock` 显示锁和 `synchronized`。总结一下,`volatile` 能以较低的成本减小整个程序脆弱性。


当然,还会有更多问题有待讨论。欢迎在下面评论,抛出你的想法。本文源代码可以在[这里][1]找到,还有一个[测试用例][2]。


希望你喜欢这篇文章。


[1]:https://github.com/nschlimm/playground-java8/tree/master/src/main/java/org/projectbarbel/playground/safeaccessor

[2]:https://github.com/nschlimm/playground-java8/blob/master/src/main/java/org/projectbarbel/playground/safeaccessor/SafeAccessorServiceTest.java


标签: 
相关文章推荐