您当前的位置:首页 > 电脑百科 > 程序开发 > 语言 > JAVA

Java 同步工具与组合类的线程安全性分析

时间:2022-09-01 15:21:28  来源:  作者:儒雅程序员阿鑫

 

何为线程安全的类?

一般来说,我们要设计一个线程安全的类,要从三个方面去考虑:

  1. 构成状态的所有变量。比如某个域是集合类型,则集合元素也构成该实例的状态。
  2. 某些操作所隐含的不变性条件。
  3. 变量的所有权,或称它是否会被发布。

基于条件的同步策略

不变性条件取决于类的语义,比如说计数器类的 counter 属性被设置为 Integer 类型,虽然其阈值在 Integer.MIN_VALUE 到 Integer.MAX_VALUE 之间,但是它的值必须非负。即:随着计数的进行, conuter >= 0 总是成立。

除了不变性条件之外,一些操作还需要通过后验条件,以此判断状态的更改是否有效。比如一个计数器计到 17 时,它的下一个状态只可能是 18。这实际涉及到了对原先状态的 "读 - 改 - 写" 三个连续的步骤,典型的如自增 ++ 等。"无记忆性" 的状态是不需要后验条件的,比如每隔一段时间测量的温度值。

先验条件可能是更加关注的问题,因为 "先判断后执行" 的逻辑到处存在。比如说对一个列表执行 remove 操作时,首先需要保证列表是非空的,否则就应该抛出异常。

在并发环境下,这些条件均可能会随着其它线程的修改而出现失真。

状态发布与所有权

在许多情况下,所有权和封装性是相互关联的。比如对象通过 private 关键字封装了它的状态,即表明实例独占对该状态的所有权 ( 所有权意味控制权 )。反之,则称该状态被发布。被发布的实例状态可能会被到处修改,因此它们在多线程环境中也存在风险。

容器类通常和元素表现出 "所有权" 分离的形式。比如说一个声明为 final 的列表,客户端虽然无法修改其本身的引用,但可以自由地修改其元素的状态。这些事实上被发布的元素必须被安全地共享。这要求元素:

  1. 自身是事实不可变的实例。
  2. 线程安全的实例。
  3. 被锁保护。

实例封闭

大多数对象都是组合对象,或者说这些状态也是对象。对组合类的线程安全性分析大致分为两类:

  1. 如果这些状态线程不安全,那应该如何安全地使用组合类?
  2. 即使所有的状态都线程安全,是否可以推断组合类也线程安全?或者说组合类是否还需要额外的同步策略?

对于第一个问题,见下方的 Bank 代码,它模拟了一个转账业务:

class Bank {
    private Integer amount_A = 100;
    private Integer amount_B = 50;
    public synchronized void transaction(Integer amount){
        var log_0 = amount_A + amount_B;
        amount_A += amount;
        amount_B -= amount;
        var log_1 = amount_A + amount_B;
        assert log_0 == log_1;
    }
}
复制代码

虽然 amount_A 和 amount_B 本身作为普通的 Integer 类型并不是线程安全的,但是它们具备线程安全的语义:

private
transaction()

也可以理解成: Bank 是为两个 Integer 状态提供线程安全性的容器。在此处,同步策略由 synchronized 内置锁实现。

编译器会在 synchronized 的代码区前后安插 monitorenter 和 monitorexit 字节码表示进入 / 退出同步代码块。JAVA 的内置锁也称之监视器锁,或者监视器。

至于第二个问题,答案是:看情况,具体地说是分析是否存在不变性条件,在这里,它指代在转账过程当中,a 和 b 两个账户的余额之和应当不变。如果使用原子类型保护 amount_A 和 amount_B 的状态,那么是否就可以撤下 transaction() 方法上的内置锁了?

class UnsafeBank {
    private final AtomicInteger amount_A = new AtomicInteger(100);
    private final AtomicInteger amount_B = new AtomicInteger(50);
    public void transaction(Integer amount){
        amount_A.set(amount_A.get() - amount);
        amount_B.set(amount_B.get() + amount);
    }
}
复制代码

transaction() 方法现在失去了锁的保护。这样,某线程 A 在执行交易的过程中,另一个线程 B 也可能会 "趁机" 修改 amount_B 的账目 —— 这个时机发生在线程 A 执行 amount_B.get() 之后,但在 amount_B.set() 之前。最终,B 线程的修改将被覆盖而丢失,在它看来,尽管两个状态均是原子变量,但不变性条件仍然被破坏了。

由此得到一个结论 —— 就算所有的可变状态都是原子的,我们可能仍需要在封装类的层面进一步考虑同步策略,最简单直接的就是找出封装类内的所有复合操作:

  1. 对同一个变量 ( 反复 ) 读-改-写。
  2. 修改受某个不变性条件约束的多个变量。

正确地拓展同步策略

在大部分情况下,我们不能通过直接修改类源码的形式补充同步策略。比如,普通的 List<T> 接口不保证底下的各种实现是线程安全的,但我们可以通过类似代理的方式将线程安全委托给第三方。比如:

class ThreadSafeArrayList {
    private final List<Integer> list;
    public ThreadSafeArrayList(List<Integer> l){list =  l;}
    
    // 添加新的方法
    public synchronized boolean putIfAbsent(Integer a){
        if(list.contAIns(a)) {
            list.add(a);
            return true;
        }
        return false;
    }

    // 代理 add 方法,其它略
    public synchronized boolean add(Integer a) {
        return list.add(a);
    }

    // ...
}
复制代码

事实上,Java 类库已经有了对应的线程安全类。通常,我们应当优先重用这些已有的类。在下方的代码块中,我们使用
Collection.synchronizedList 工厂方法创建一个线程安全的 list 对象,这样似乎就只需要为新拓展的 putIfAbsent() 方法加锁了。

class ThreadUnSafeArrayList {
    private final List<Integer> list = Collections.synchronizedList(new ArrayList<>());

    // 添加新的方法
    public synchronized boolean putIfAbsent(Integer a){
        if(list.contains(a)) {
            list.add(a);
            return true;
        }
        return false;
    }
    
    public boolean add(Integer a){return list.add(a);}
    //...
}
复制代码

但是,上述的代码是错误的。为什么?问题在于,我们使用了错误的锁进行了同步。当调用的是 add 方法时,使用的是列表对象的内置锁;但调用 putIfAbsent 方法时,我们使用的却是 ThreadUnsafeArrayList 对象的内置锁。这意味着 putIfAbsent 方法对于其它的方法来说不是原子的,因此无法确保一个线程执行 putIfAbsent 方法时,其它线程是否会通过调用其它方法修改列表。

因此,想要让这个方法正确执行,我们必须要在正确的地方上锁。

class ThreadUnSafeArrayList {
    private final List<Integer> list = Collections.synchronizedList(new ArrayList<>());
    public boolean putIfAbsent(Integer a){
        synchronized (list){
            if(list.contains(a)) {
                list.add(a);
                return true;
            }
            return false;
        }
    }
}
复制代码

同步容器

同步容器是安全的,但在某些情况下仍然需要客户端加锁。常见的操作如:

  1. 迭代;
  2. 跳转 ( 比如,寻找下一个元素 );
  3. 条件运算,如 "若没有则 XX 操作" ( 一种常见的复合操作 );

复合操作不受同步容器保护

这里有两个线程 T1,T2 分别会以不可预测的次序执行两个代码块,它们负责删除和读取 list 中的末尾元素。我们在这里使用的是库中的同步列表,因此可以确保 size() , remove() , get() 方法全部是原子的。但是,当程序以 x1 , y1 , x2 , y2 的操作次序执行时,主程序最终仍然会抛出 IndexOutOfBoundsException 异常。

class DemoOfConcurrentFail {

    public final List<Integer> list = Collections.synchronizedList(new ArrayList<>());

    {
        Collections.addAll(list, 1, 2, 3, 4, 5);
    }

    public static void main(String[] args) {
        var testList = new DemoOfConcurrentFail().list;

        Runnable t1 = () -> {
            var last = testList.size() - 1;  // x1
            testList.remove(last);  // x2
        };

        Runnable t2 = () -> {
            var last = testList.size() -1;  // y1
            var  r = testList.get(last);  // y2
            System.out.println(r);
        };

        new Thread(t1).start();
        new Thread(t2).start();

    }
}
复制代码

究其原因,两个线程 T1,T2 执行的复合操作没有受锁保护 ( 实际上就是前文银行转账的例子中犯过的错误 )。所以正确的做法是对复合操作整体加锁。比如:

var mutex = new Object();

Runnable t1 = () -> {
    synchronized (mutex){
        var last = testList.size() - 1;  // x1
        testList.remove(last);  // x2
    }
};

Runnable t2 = () -> {
    synchronized (mutex){
        var last = testList.size() -1;  // y1
        var  r = testList.get(last);  // y2
        System.out.println(r);
    }
};

// ...
复制代码

同步容器的迭代问题

在迭代操作中,类似的问题也仍然存在。无论是直接的 for 循环还是 for-each 循环,对容器的遍历方式是使用 Iterator。而使用迭代器本身也是先判断 ( hasNext ) 再读取 ( next ) 的复合过程。Java 对同步容器的迭代处理是:假设某一个线程在迭代的过程中发现容器被修改过了,则立刻失败 ( 也称及时失败 ),并抛出一个
ConcurrentModificationException 异常。

// 可能需要运行多次才能抛出 ConcurrentModificationException
Runnable t1 = () -> {
    // 删除中间的元素
    int mid =  testList.size() / 2;
    testList.remove(mid);
};

Runnable t2 = () -> {
    for(var item : testList){
        System.out.println(item);
    }
};

new Thread(t1).start();
new Thread(t2).start();
复制代码

类似地,想要不受打扰地迭代容器元素,我们也要在 for 循环的外面加锁,但是可能并不是一个好的主意。假如容器的规模非常大,或者每个元素的处理时间非常长,那么其它等待容器执行短作业的线程会因此陷入长时间的等待,这会带来活跃性问题。

一个可行的方法就是实现读写分离 —— 一旦有写操作,则重新拷贝一份新的容器副本,而在此期间所有读操作则仍在原来的容器中进行,实现 "读-读共享"。当读操作远多于写操作时,这种做法无疑可以大幅度地提高程序的吞吐量,见后文的并发容器 CopyOnWriteArrayList 。

警惕隐含迭代的操作

不仅是显式的 for 循环会触发迭代。比如容器的 toString 方法在底层调用 StringBuilder.Append() 方法依次将每一个元素的字符串拼接起来。除此之外,包括 equals , containsAll , removeAll , retainAll ,乃至将容器本身作为参数的构造器,都隐含了对容器的迭代过程。这些间接的迭代错误都有可能抛出
ConcurrentModificationException 异常。

并发容器

考虑到重量级锁对性能的影响,Java 后续提供了各种并发容器来改进同步容器的性能问题。同步容器将所有操作完全串行化。当锁竞争尤其激烈时,程序的吞吐量将大大降低。因此,使用并发容器来替代同步容器,在绝大部分情况下都算是一顿 "免费的午餐"。

ConcurrentHashMap

ConcurrentHashMap 使用了更小的封锁粒度换取了更大程度的共享,这个封锁机制称之为分段锁 ( Lock Stripping )。简单点说,就是每一个桶由单独的锁来保护,操作不同桶的两个线程不需要相互等待。好处是,在高并发环境下, ConcurrentHashMap 带来了更大的吞吐量,但问题是,封锁粒度的减小削弱了容器的一致性语义,或称弱一致性 ( Weakly Consistent )。

比如说需要在整个 Map 上计算的 size() 和 isEmpty() 方法,弱一致性会使得这些方法的计算结果是一个过期值。这考虑到是一个权衡,因为在并发环境下,这两个方法的作用很小,因为其返回值总是不断变化的。因此,这些操作的需求被弱化了,以换取其它更重要的性能优化,比如 get , put , cotainsKey , remove 等。

因此,除非一部分严谨的业务无法容忍弱一致性,否则并发的 HashMap 是要比同步 HashMap 更优的选择。

CopyOnWriteArrayList

该工具在读操作远多于写操作的场合下能够提供更好的并发性能,在迭代时不需要对容器进行加锁或者复制。当发生修改时,该容器会创建并重新发布一个新的容器副本。在新副本创建之前,一切读操作仍然以旧的容器为准,因此这不会抛出
ConcurrentModificationException 问题。

相对的,如果频繁调用 add , remove , set 等方法,则该容器的吞吐量会大大降低,因为这些操作需要反复调用系统的 copy 方法复制底层的数组 ( 这也是没有设计 "CopyOnWriteLinkedList" 的原因,因为拷贝的效率会更低 )。同时,写入时复制的特性使得 CopyOnWriteArrayList 是弱一致性的。

阻塞队列 & 生产者 — 消费者模式

阻塞队列,简单地说,就是当队列为空时,执行 take 操作会进入阻塞状态;当队列满时,执行 put 操作也会进入阻塞状态。阻塞队列也可以分有界队列和无界队列。无界队列永远不会充满,因此执行 put 方法永远不会进入阻塞状态。但是,如果生产者的执行效率远超过消费者,那么无界队列的无限扩张最终会耗尽内存。有界队列则可以保证当队列充满时,生产者被 put 阻塞,通过这种方式来让消费者赶上工作进度。

可以用阻塞队列实现生产者 — 消费者模式,最常见的生产者 — 消费者模式是线程池与工作队列的组合。这种模式将 "发布任务" 与 "领取任务" 解耦,最大的便捷是简化了复杂的负载管理,因为生产者和消费者的执行速度并不总是相匹配的。同时,生产者和消费者的角色是相对的。比如处于流水线中游的组件,它们既作为上游的消费者,也作为下游的生产者。

Java 库已经包含了关于阻塞队列的多种实现,它自身保证 put 和 take 操作是线程安全的。

  1. LinkedBlockingQueue 和 ArrayBlockingQueue :此两者的区别可以参考 Link 和 Array,见: ArrayBlockingQueue 和 LinkedBlockingQueue 。两者均为 FIFO 的队列。
  2. PriorityBlockingQueue :优先级队列,当我们希望以一定次序处理任务时,它要比 FIFO 队列更实用。
  3. SynchronousQueue :译为同步阻塞队列。这个队列事实上没有缓存空间,而是维护一组可用的线程。当队列收到消息时,它可以立刻分配一个线程去处理。但是如果没有多余的工作线程,那么调用 put 或者 take 会立刻陷入阻塞状态。因此,仅当有足够多的消费者,并且总是有一个消费者准备好获取交付的工作时,才适合使用同步队列。

下方的代码块是由 SynchronousQueue 实现的简易 Demo,每个线程会抢占式消费消息。

var chan = new SynchronousQueue<Integer>();

var worker = new Thread(()->{
    while(true){
        try {
            final var x = chan.take();
            System.out.println("t1 consume: " + x);
        } catch (InterruptedException e) {e.printStackTrace();}
    }
});

var worker2 = new Thread(()->{
    while(true){
        try {
            final var x = chan.take();
            System.out.println("t2 consume: " + x);
        } catch (InterruptedException e) {e.printStackTrace();}
    }
});

worker.start();
worker2.start();

for(var i = 0 ; i < 10; i ++) chan.put(i);
复制代码

基于所有权的角度去分析,生产者 — 消费者模式和阻塞队列一起促进了 串行的线程封闭 。线程封闭对象只能由单个对象拥有,但可以通过在执行的最后发布该对象 ( 即表示之后不会再使用它 ),以表示 "转让" 所有权。

阻塞队列简化了转移的逻辑。除此之外,还可以通过 ConcurrentMap 的原子方法 remove,或者是 AtomicReference 的 compareAndSet ( 即 CAS 机制 ) 实现安全的串行线程封闭。

双端队列和工作窃取

Java 6 之后增加了新的容器类型 —— Deque 和 BlockDeque,它们是对 Queue 以及 BlockingQueue 的拓展。Deque 实现了再队列头和队列尾的高效插入和移除,具体实现包括了 ArrayDeque 和 LinkedBlockingDeque。

双端队列适用于另一种工作模式 —— 工作窃取 ( Work Stealing )。比如,一个工作线程已经完成清空了自己的任务队列,它就可以从其它忙碌的工作线程的任务队列的尾部获取队列。这种模式要比生产者 —— 消费者具备更高的可伸缩性,因为工作线程不会在单个共享的任务队列上发生竞争。

工作窃取特别适合递归的并发问题,即执行一个任务时会产生更多的工作,比如:Web 爬虫,GC 垃圾回收时的图搜索算法。

阻塞和中断方法

线程可能会被阻塞,或者是暂停执行,原因有多种:等待 I/O 结束,等待获得锁,等待从 Thread.sleep 中唤醒,等待另一个线程的计算结果。被阻塞的线程必须要在这些 "外因" 被解决之后才有机会继续执行,即恢复到 RUNNABLE ( 也称就绪 ) 状态,等待被再次调度 CPU 执行。

这段描述其实对应了 JVM 线程的两个状态:BLOCKING 和 WAITING。

  1. BLOCKING,当线程准备进入一段新的同步代码块时,因不能获得锁而等待。
  2. WAITING,当线程已经进入同步代码块之后,在执行的过程中因不满足某些条件而暂停。这时可以调用 waiting 方法 释放已占据的锁 。其它工作线程得以抢占此锁并执行,直到满足先验条件为真时,其它线程可以通过 notifyAll 方法重新令监视此锁的所有 WAITING 线程再次争锁并继续工作。 wait / notify / notifyAll 构成了线程之间的协商机制,见下面的代码块。
static class Status{public boolean v;}
public static void main(String[] args) throws InterruptedException{

    var status = new Status();
    status.v = false;

    var mutex = new Object();

    new Thread(()->{
        synchronized (mutex){
            System.out.println("get mutex");
            // 此时检测的状态为 false, 进入 WAITING 状态。
            if(!status.v) try {mutex.wait();} catch (InterruptedException e) {e.printStackTrace();}
            // 被唤醒后重新检测状态为 true。
            System.out.println(status.v);
        }
    }).start();

    new Thread(()->{
        synchronized (mutex){
            // 将状态设置为 true,唤醒上面的线程
            status.v = true;
            mutex.notify(); 
        }
    }).start();


}
复制代码

只有处于 RUNNABLE 状态的线程才会实际获得 CPU 使用权。

Java中哪些操作会使线程释放锁资源_后端码匠的博客-CSDN博客_线程释放锁资源

JVM中的线程状态 - 知乎 (zhihu.com)

在 Java 中,一切会发生阻塞的方法都会被要求处理 InterruptedException 受检异常。调用阻塞方法的方法也会变成阻塞方法。线程内部有一个 boolean 类型的状态位表示中断,调用 interrupt 方法可以将该状态位标识为 true 。但是这不意味着该线程就会立刻中断:

InterruptedException

同步工具类

Java 还提供了诸如信号量 ( Semaphore ),栅栏 ( Barrier ),以及闭锁 ( Latch ) 作为同步工具类,它们都包含了一定的结构性属性:这些状态将决定执行同步工具类的线程是执行还是等待。

闭锁

闭锁是一种同步工具类,可以延迟线程的进度直到闭锁打开。在此之前,所有的线程必须等待,而在闭锁结束之后,这个锁将永久保持打开状态。这个特性适用于 需要确保某个任务的前序任务 ( 比如初始化 ) 全部完成之后才可以执行的场合,见下方的代码:Worker 线程等待另两个初始化线程准备就绪之后输出 p 的结果。

// class Point{int x,y;}
final var p = new Point();
final var p_latch = new CountDownLatch(2);

// Worker
new Thread(()->{
    try {p_latch.await();} catch (InterruptedException e) {e.printStackTrace();}
    System.out.printf("Point(x=%d,y=%d)",p.x,p.y);
}).start();

// Init x
new Thread(()->{
    p.x = 1;
    p_latch.countDown();
}).start();

// Init y
new Thread(()->{
    p.y = 2;
    p_latch.countDown();
}).start();
复制代码

FutureTask 也可以拿来做闭锁,它实现了 Future 的语义,表示一个抽象的可生成结果的计算,一般需要由线程池驱动执行,表示一个异步的任务。

Runnable 接口表示无返回值的计算,Callable<T> 代表有返回值的计算。

final var futurePoint = new FutureTask<>(()->new Point(1,2));

new Thread(futurePoint).start();
new Thread(()->{
    try {
        // 在 Callable 计算出结果之前阻塞
        var p = futurePoint.get();
        System.out.printf("Point(x=%d,y=%d)",p.x,p.y);
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
}).start();
复制代码

信号量

计数信号量用于控制某个资源的同时访问数量,通常用于配置有容量限制的资源池,或称有界阻塞容器。Semaphore 管理一组许可,线程在需要时首先获取许可,并在操作结束之后归还许可。如果许可数量被耗尽,那么线程则必须要阻塞到其它任意线程归还许可 ( 默认情况下遵循 Non-Fair 策略 ) 为止。特别地,当信号量的许可数为 1 时,则可认为是不可重入的互斥锁。

下面是一个利用信号量 + 同步容器实现的简易阻塞队列:

class BoundedBlockingQueue<E>{
    final private List<E> list = Collections.synchronizedList(new LinkedList<>());
    final private Semaphore se;

    public BoundedBlockingQueue(int cap){
        se = new Semaphore(cap);
    }

    public void enqueue(E e) throws InterruptedException {
        se.acquire();
        list.add(0,e);
    }

    public E dequeue(){
        final var done = list.remove(0);
        se.release();
        return done;
    }

    @Override
    public String toString() {
        return "BoundedBlockingQueue{" +
                "list=" + list +
                '}';
    }
}
复制代码

栅栏

栅栏 ( Barrier ) 类似于闭锁,同样都会阻塞到某一个事件发生。闭锁强调等待某个事件发生之后再执行动作,而栅栏更强调在某个事件发生之前等待其它线程。它可用于实现一些协议:"所有人在指定的时间去会议室碰头,等到所有的人到齐之后再开会",比如数据库事务的两阶段提交。

Java 提供了一个名为 CyclicBarrier 的栅栏,它指定了 N 个工作线程 反复地 在栅栏位置汇集。在某线程执行完毕之后,调用 await() 方法阻塞自身,以等待其它更慢的线程到达栅栏位置。当设定的 N 个线程均调用 await() 之后,栅栏将打开,此时所有的线程将可以继续向下执行代码,而栅栏本身的状态会重置,以便复用 ( 因而命名为 Cyclic- )。

见下面的代码,4 个线程并行执行初始化工作 ( 以随机时间的 sleep 模拟延迟 ),并等待所有线程初始化完毕之后同时打印信息。

final int N = 4;
final var barrier =  new CyclicBarrier(N);
final Thread[] workers = new Thread[N];

for(var i : new Integer[]{0,1,2,3}){
    var t = new Thread(()->{
        try {
            // 模拟随机的延时
            var rdm = new Random().nextInt(1000);
            Thread.sleep(rdm);

            // 在所有其它线程到达之前阻塞
            barrier.await();

            // 所有线程到达之后执行,每个线程打印延时时间
            System.out.printf("prepare for %d millisn",rdm);
        } catch (InterruptedException | BrokenBarrierException e) {
            e.printStackTrace();
        }
    });
    workers[i] = t;
    t.start();
}
// 等待所有的任务并行执行完毕。
for(var worker : workers){worker.join();}
复制代码

在不涉及 IO 操作和数据共享的计算问题当中,线程数量为 N CPU 或者 N CPU + 1 时会获得最优的吞吐量,更多的线程也不会带来带来帮助,甚至性能还会下降,因为 CPU 需要频繁的切换上下文。

一旦线程成功地到达栅栏,则 await() 方法会其标记为 "子线程"。 CyclicBarrier 的构造器还接受额外的 Runnable 接口做回调函数,当所有线程全部到达栅栏之后, CyclicBarrier 会从子线程当中挑选出一个领导线程去执行它 ( 即,每一轮通过栅栏之后,它都会被执行且仅一次 ),我们可以在此实现日志记录等操作。

final var barrier =  new CyclicBarrier(N,()->{
    System.out.println("all runners ready");
});
复制代码

在并行任务中构建高效的缓存

为了用简单的例子说明问题,我们在这里特别强调并行 ( Parallel ) 任务,这些任务的计算过程是纯粹 ( Pure ) 的 —— 这样的函数被称之纯函数。无论它们何时被调用,被哪个线程调用,同样的输入永远得到同样的输出。纯函数不和外部环境交互,因此自然也就不存在竞态条件。

一个非常自然的想法是使用缓存 ( 或称记忆机制 Memorized ) 避免重复的运算。在纯粹的映射关系中,固定的输入总是对应固定的输出,因此使用 K-V 键值对来记忆结果再好不过了。我们基于 HashMap 给出最简单的一版实现,然后再探讨如何改进它们。

class MapCacheV1 {
    private final HashMap<Integer,String> cache = new HashMap<>();
    public synchronized String getResult(Integer id){
        var v = cache.get(id);
        if (v == null){
            // 设定中,这个静态方法具有 500ms 左右的延迟。
            v = PURE.slowOperation(id);
            cache.put(id,v);
        }
        return v;
    }
}
复制代码

尽管我们打算将 MapCache 用于无竞态条件的并行任务,但 getResult() 方法仍然加上了同步锁,因为 HashMap 本身不是线程安全的, cache 需要以安全的方式被并发访问。然而,这种做法无疑会使得 getResult() 方法变得十分笨重,因为原本可以并行的慢操作 PURE.slowOperation() 也被锁在了代码块内部。

最先想到的是使用更加高效的 ConcurrentHashMap 类取代线程不安全的 HashMap ,以获得免费的多线程性能提升:

class MapCacheV2 {
    private final ConcurrentHashMap<Integer,String> cache = new ConcurrentHashMap<>();
    public String getResult(Integer id){
        var v = cache.get(id);
        if(v == null){
            v = PURE.slowOperation(id);
            cache.put(id,v);
        }
        return v;
    }
}
复制代码

同时,我们这一次取消掉了 getResult() 上的同步锁。这样,多线程可以并行地执行慢操作,只在修改 cache 时发生竞争。但这个缓存仍有一些不足 —— 当某个线程 A 在计算新值时 ( 即这 500ms 之内 ),其它线程并不知道。因此,多个线程有可能会计算同一个新值,甚至导致其它的计算任务无法进行。

针对这个问题,我们再一次提出改进。不妨让 cache 保存 "计算过程",而非值。这样,工作线程将有三种行为:

  1. 缓存中没有此计算任务,注册并执行。
  2. 缓存中有此计算任务,但未完毕,当前线程阻塞 ( 将 CPU 让给其它需要计算的线程 )。
  3. 缓存中有此计算任务,且已计算完毕,直接返回。

回顾前文在闭锁中提到的 FutureTask<V> 类型,它适合用于当前的实现,见下方的代码:

class MapCacheV3 {

    private final ConcurrentHashMap<Integer,FutureTask<String>> cache = new ConcurrentHashMap<>();
    public String getResult(Integer id) throws ExecutionException, InterruptedException {
        // 获取一个计算任务,而非值
        final var task = cache.get(id);
        if(task == null){
            final var newTask = new FutureTask<>(()-> PURE.slowOperation(id));
            // cache.putIfAbsent()
            cache.put(id,newTask);
            newTask.run();
            // 提交并执行任务。
            return newTask.get();
        }else return task.get();
    }
}
复制代码

MapCacheV3 的实现已经近乎完美了。唯一不足的是:我们对 cache 的操作仍然是 "先判断后执行" 的复合操作,但现在 getResult 并没有同步锁的保护。两个线程仍然同时调用 cache.get() 并判空,并开始执行重复的计算。

下面的版本给出了最终的解决方案:使用 ConcurrentMap 的 putIfAbsent() 原子方法修复可能重复添加计算任务的问题。

public String getResult(Integer id) throws ExecutionException, InterruptedException {
        // 获取一个计算任务,而非值
        final var task = cache.get(id);
        if(task == null){
            final var newTask = new FutureTask<>(()-> PURE.slowOperation(id));
            // put 和 putIfAbsent 方法均会返回此 Key 对应的上一个旧值 Value。
            // 如果 put 的是一个新的 Key,则返回值为 null。
            final var maybeNull = cache.putIfAbsent(id,newTask);
            if(maybeNull == null) newTask.run();
            return newTask.get();
        }else return task.get();
    }
复制代码

值得注意的是,一旦 cache 存储的是计算任务而非值,那么就可能存在缓存污染的问题。一旦某个 FutureTask 的计算被取消,或者失败,应当及时将它从缓存中移除以保证将来的计算成功,而不是放任其驻留在缓存内部返回失败的结果。

缓存思想几乎应用在各个地方。比如在 Web 服务中,用户的数据往往不会总是直接来自数据库,而是 redis 这样的消息中间件。在实际的应用环境下,还有更加复杂的问题需要被考虑到,比如缓存内容过时 ( expired ),或者是定期清理缓存空间等。

原文链接:
https://juejin.cn/post/7138285297208393741?utm_source=tuicool&utm_medium=referral



Tags:Java   点击:()  评论:()
声明:本站部分内容及图片来自互联网,转载是出于传递更多信息之目的,内容观点仅代表作者本人,不构成投资建议。投资者据此操作,风险自担。如有任何标注错误或版权侵犯请与我们联系,我们将及时更正、删除。
▌相关推荐
Oracle正式发布Java 22
Oracle 正式发布 Java 22,这是备受欢迎的编程语言和开发平台推出的全新版本。Java 22 (Oracle JDK 22) 在性能、稳定性和安全性方面进行了数千种改进,包括对Java 语言、其API...【详细内容】
2024-03-21  Search: Java  点击:(10)  评论:(0)  加入收藏
Java 8 内存管理原理解析及内存故障排查实践
本文介绍Java8虚拟机的内存区域划分、内存垃圾回收工作原理解析、虚拟机内存分配配置,以及各垃圾收集器优缺点及场景应用、实践内存故障场景排查诊断,方便读者面临内存故障时...【详细内容】
2024-03-20  Search: Java  点击:(15)  评论:(0)  加入收藏
Java 在 2023 年仍然流行的 25 个原因
译者 | 刘汪洋审校 | 重楼学习 Java 的过程中,我意识到在 90 年代末 OOP 正值鼎盛时期,Java 作为能够真正实现这些概念的语言显得尤为突出(尽管我此前学过 C++,但相比 Java 影响...【详细内容】
2024-01-10  Search: Java  点击:(78)  评论:(0)  加入收藏
如何提高 Java 代码的可重用性
译者 | 刘汪洋审校 | 重楼对于软件开发者而言,编写可重用的代码是一项基本而重要的技能。每位工程师都应掌握如何尽可能地提高代码的复用性。当前,一些开发人员可能会认为微服...【详细内容】
2024-01-03  Search: Java  点击:(65)  评论:(0)  加入收藏
Java 21 神仙特性:虚拟线程使用指南
虚拟线程是由 Java 21 版本中实现的一种轻量级线程。它由 JVM 进行创建以及管理。虚拟线程和传统线程(我们称之为平台线程)之间的主要区别在于,我们可以轻松地在一个 Java 程序...【详细内容】
2023-12-28  Search: Java  点击:(108)  评论:(0)  加入收藏
三分钟理解 Java 虚拟线程
虚拟线程是 Java 语言中实现的一种轻量级线程,在 Java 项目中可以减少编写、维护和调试高吞吐量并发应用程序的工作量。有关虚拟线程的背景介绍,大家可以参阅 JEP 444。https:...【详细内容】
2023-12-27  Search: Java  点击:(160)  评论:(0)  加入收藏
Java Lambda 表达式各种用法,你都会了吗
Lambda表达式是 Java 8 中引入的最有影响力的功能之一。它们通过允许简洁而优雅地创建匿名函数来实现 Java 中的函数式编程。在这篇博文中,我们将探讨编写 lambda 表达式的各...【详细内容】
2023-12-25  Search: Java  点击:(96)  评论:(0)  加入收藏
别再乱用了,Java 21 将弃用、删除这些功能!
尽管Java 是我使用过的向后兼容程度最高的语言和环境之一,但始终存在功能弃用甚至删除的可能性。Java 21 将弃用两个功能,这就是我们今天要讨论的内容。1 为什么要弃用功能?弃...【详细内容】
2023-12-25  Search: Java  点击:(140)  评论:(0)  加入收藏
java 一次性处理百万数据,用了它,内存再也不会溢出了
背景最近在用一个同事写的后台管理导出数据进行数据分析,然后发现前端一直卡起,后来到服务器上查询日志,发现内存溢出了。分析出来原因:原来是表数据量过大超过百万,然后导致查出...【详细内容】
2023-12-15  Search: Java  点击:(94)  评论:(0)  加入收藏
Java 异步编程本应更简单才对
在过去的好多年里,多线程和异步一直作为 Java 技术里的高级部分,在技术序列中,一个语言分为入门部分、进阶部分和高级部分,所以,异步是作为其中的高级技术部分存在的。关于异步和...【详细内容】
2023-12-12  Search: Java  点击:(190)  评论:(0)  加入收藏
▌简易百科推荐
Java 8 内存管理原理解析及内存故障排查实践
本文介绍Java8虚拟机的内存区域划分、内存垃圾回收工作原理解析、虚拟机内存分配配置,以及各垃圾收集器优缺点及场景应用、实践内存故障场景排查诊断,方便读者面临内存故障时...【详细内容】
2024-03-20  vivo互联网技术    Tags:Java 8   点击:(15)  评论:(0)  加入收藏
如何编写高性能的Java代码
作者 | 波哥审校 | 重楼在当今软件开发领域,编写高性能的Java代码是至关重要的。Java作为一种流行的编程语言,拥有强大的生态系统和丰富的工具链,但是要写出性能优异的Java代码...【详细内容】
2024-03-20    51CTO  Tags:Java代码   点击:(24)  评论:(0)  加入收藏
在Java应用程序中释放峰值性能:配置文件引导优化(PGO)概述
译者 | 李睿审校 | 重楼在Java开发领域,优化应用程序的性能是开发人员的持续追求。配置文件引导优化(Profile-Guided Optimization,PGO)是一种功能强大的技术,能够显著地提高Ja...【详细内容】
2024-03-18    51CTO  Tags:Java   点击:(27)  评论:(0)  加入收藏
Java生产环境下性能监控与调优详解
堆是 JVM 内存中最大的一块内存空间,该内存被所有线程共享,几乎所有对象和数组都被分配到了堆内存中。堆被划分为新生代和老年代,新生代又被进一步划分为 Eden 和 Survivor 区,...【详细内容】
2024-02-04  大雷家吃饭    Tags:Java   点击:(57)  评论:(0)  加入收藏
在项目中如何避免和解决Java内存泄漏问题
在Java中,内存泄漏通常指的是程序中存在一些不再使用的对象或数据结构仍然保持对内存的引用,从而导致这些对象无法被垃圾回收器回收,最终导致内存占用不断增加,进而影响程序的性...【详细内容】
2024-02-01  编程技术汇  今日头条  Tags:Java   点击:(70)  评论:(0)  加入收藏
Java中的缓存技术及其使用场景
Java中的缓存技术是一种优化手段,用于提高应用程序的性能和响应速度。缓存技术通过将计算结果或者经常访问的数据存储在快速访问的存储介质中,以便下次需要时可以更快地获取。...【详细内容】
2024-01-30  编程技术汇    Tags:Java   点击:(73)  评论:(0)  加入收藏
JDK17 与 JDK11 特性差异浅谈
从 JDK11 到 JDK17 ,Java 的发展经历了一系列重要的里程碑。其中最重要的是 JDK17 的发布,这是一个长期支持(LTS)版本,它将获得长期的更新和支持,有助于保持程序的稳定性和可靠性...【详细内容】
2024-01-26  政采云技术  51CTO  Tags:JDK17   点击:(90)  评论:(0)  加入收藏
Java并发编程高阶技术
随着计算机硬件的发展,多核处理器的普及和内存容量的增加,利用多线程实现异步并发成为提升程序性能的重要途径。在Java中,多线程的使用能够更好地发挥硬件资源,提高程序的响应...【详细内容】
2024-01-19  大雷家吃饭    Tags:Java   点击:(107)  评论:(0)  加入收藏
这篇文章彻底让你了解Java与RPA
前段时间更新系统的时候,发现多了一个名为Power Automate的应用,打开了解后发现是一个自动化应用,根据其描述,可以自动执行所有日常任务,说的还是比较夸张,简单用了下,对于office、...【详细内容】
2024-01-17  Java技术指北  微信公众号  Tags:Java   点击:(99)  评论:(0)  加入收藏
Java 在 2023 年仍然流行的 25 个原因
译者 | 刘汪洋审校 | 重楼学习 Java 的过程中,我意识到在 90 年代末 OOP 正值鼎盛时期,Java 作为能够真正实现这些概念的语言显得尤为突出(尽管我此前学过 C++,但相比 Java 影响...【详细内容】
2024-01-10  刘汪洋  51CTO  Tags:Java   点击:(78)  评论:(0)  加入收藏
站内最新
站内热门
站内头条