java多线程实用库

July 18, 2022

多线程工具类由 java.util.concurrent 包提供
锁工具类由 java.util.concurrent.locks 包提供

java.util.concurrent.locks 包：https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/locks/ReentrantLock.html
锁接口：https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/locks/Lock.html

ReentrantLock 类，可重入锁

ReentrantLock 锁由 java.util.concurrent.locks 包提供，可作为 synchronized 语法的替代

java

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
class X {
    private final Lock lock = new ReentrantLock();

    public void m() {
        lock.lock();  // 加锁
        try {
            // ... method body
        } finally {
            lock.unlock(); // 解锁
        }
    }
}

ReentrantLock（而非 Lock 接口）还提供更多实例方法。
参见 API：https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/locks/ReentrantLock.html

Lock 接口

https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/locks/Lock.html

Lock 接口还允许尝试获取锁

java

// lock.lockInterruptibly(); // 尝试获取锁除非当前线程被 interrupted
// lock.tryLock(1, TimeUnit.SECONDS); // 指定1s
if (lock.tryLock()) { // 尝试获取锁
    try {
        // ...
    } finally {
        lock.unlock();
    }
}

Condition 类，实现 wait/notify

https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/locks/Condition.html

java

import java.util.*;
import java.util.concurrent.locks.*;

class TaskQueue {
    private final Lock lock = new ReentrantLock();
    private final Condition condition = lock.newCondition(); // 注意这个 Condition 对象是绑定到 Lock 对象上的
    private final LinkedList<String> queue = new LinkedList<>();

    public void addTask(String s) {
        lock.lock();
        try {
            queue.add(s);
            condition.signalAll(); // 相当于 notifyAll， signal 则相当于 notify
        } finally {
            lock.unlock();
        }
    }

    public String getTask() throws InterruptedException {
        lock.lock();
        try {
            while (queue.isEmpty()) {
                condition.await(); // 相当于 wait，此方法可能抛出 InterruptedException，因为在等待期间可能触发 interrupt()
            }
            return queue.remove();
        } finally {
            lock.unlock();
        }
    }
}

ReadWriteLock

https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/locks/ReadWriteLock.html

ReadWriteLock 是一个接口，提供了 readLock 和 writeLock 两个实例方法（返回 Lock 接口）。
可实现一个资源可以被多个线程同时读，或者被一个线程写，但是不能同时存在读和写线程。

ReentrantReadWriteLock 可重入读写锁实现了 ReadWriteLock 接口

java

import java.util.*;
import java.util.concurrent.locks.*;

class MyReadWriteMap {
    private volatile Map<String, String> map = new HashMap<>();
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    public void put(String key, String value) {
        lock.writeLock().lock(); // 加写锁
        try {
            map.put(key, value);
        } finally {
            lock.writeLock().unlock(); // 释放写锁
        }
    }

    public String get(String key) {
        lock.readLock().lock(); // 加读锁
        try {
            return map.get(key);
        } finally {
            lock.readLock().unlock(); // 释放读锁
        }
    }
}

StampedLock

<https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/locks/StampedLock.html >

StampedLock 在 ReentrantReadWriteLock 的基础上新增一种乐观读的模式，允许获取写锁后写入。
（StampedLock 是提供的是独立的接口，并非 ReadWriteLock 接口实现，即使其 API 有相似性）

乐观读模式（没有加锁）为了防止读写过程中资源不同步，还需要额外的判断。
此外，StampedLock 也同样提供独占写锁和悲观读锁 （就像 ReadWriteLock）
并且，StampedLock 提供转换为读锁、写锁、读写锁的方法

java

import java.util.concurrent.locks.*;

class Complex {
    private final StampedLock stampedLock = new StampedLock();
    private double x = 0;
    private double y = 0;

    public void set(double newX, double newY) {
        long stamp = stampedLock.writeLock(); // 获取写锁
        try {
            x = newX;
            y = newY;
        } finally {
            stampedLock.unlockWrite(stamp); // 释放写锁
        }
    }

    public double[] getAsArray() {
        long stamp = stampedLock.tryOptimisticRead(); // 获得一个乐观读锁
        double currentX = x;
        double currentY = y;

        if (!stampedLock.validate(stamp)) { // 检查乐观读锁后是否有其他写锁发生
            stamp = stampedLock.readLock(); // 获取一个悲观读锁
            try {
                currentX = x;
                currentY = y;
            } finally {
                stampedLock.unlockRead(stamp); // 释放悲观读锁
            }
        }
        return new double[]{currentX, currentY};
    }
}

线程安全集合类

java.util.concurrent 包提供了一些线程安全的集合类，实现了集合类接口，可在多线程时使用。

例如：ConcurrentMap<K,V> ConcurrentHashMap<K,V> ConcurrentLinkedQueue<E> ConcurrentLinkedDeque<E> CopyOnWriteArraySet<E> CopyOnWriteArrayList<E> 等等。

需要时参阅 API 文档即可。
https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/package-summary.html

原子操作 java.util.concurrent.atomic 包

java.util.concurrent.atomic 包提供了基于原子操作的封装类，以一种自旋锁的方式进行读写

java

var i = new AtomicInteger(1); // 可指定初始值，不指定则为默认值
System.out.println(i.get()); // 1
System.out.println(i.addAndGet(9)); // 10
System.out.println(i.incrementAndGet()); // 11

Future 类，线程返回值

Future 接口：https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/Future.html FutureTask 类：https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/FutureTask.html

java

FutureTask task = new FutureTask(new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        return 123;
    }
});
Thread t1 = new Thread(task);
t1.start();
System.out.println(task.get()); // 123

注意到线程池 ExecutorService 接口的 submit 方法还有一个参数为 Callable 的重载方法，该方法返回 Future<?>

java

ExecutorService executor = Executors.newFixedThreadPool(4);
Callable<Integer> task = new Callable<Integer>() {
    @Override
    public Integer call() throws Exception {
        return 123;
    }
};
Future<Integer> future = executor.submit(task);
Integer result = future.get();
System.out.println(result); // 1

CompletableFuture 类，异步返回

https://docs.oracle.com/en/java/javase/17/docs/api/java.base/java/util/concurrent/CompletableFuture.html

java

CompletableFuture<Integer> cf = CompletableFuture.supplyAsync(() -> 123);
cf = cf.thenApplyAsync(n -> n - 23); // 类似于 Promise.then 回调参数是上一次执行的返回值
cf.thenAccept(System.out::println); // 100，删掉上面一行则是 123
cf.exceptionally((e) -> {
    e.printStackTrace();
    return null;
});

CompletableFuture<Object> cfs = CompletableFuture.anyOf(CompletableFuture.supplyAsync(() -> 123), CompletableFutursupplyAsync(() -> 456));
cfs.thenAccept(System.out::println); // 返回的是两个CompletableFuture中先完成的那个


// 主线程不要立刻结束，否则CompletableFuture默认使用的线程池会立刻关闭:
Thread.sleep(200);

ThreadLocal

ThreadLocal 为变量在每个线程中创建了一个副本，这样每个线程都可以访问自己内部的副本变量

java

ThreadLocal<Integer> local = new ThreadLocal<>();
IntStream.range(0, 5).forEach(i -> new Thread(() -> {
    try {
        local.set(i);
        System.out.println("[" + Thread.currentThread().getName() + "]: " + local.get());
    } finally {
        local.remove();
        // 如果不进行清除，当这个线程再次执行时，仍会获取到上次设置的变量
        // 本例中由于每个线程都对local进行设置，所以也可以不用清除。

        // 原理类似于：
        // var tlm = new Map<Thread, T>;              // var local = new ThreadLocal<T>()
        // tlm.set(Thread.currentThread(), value);    // local.set(value);
        // tlm.get(Thread.currentThread());           // local.get();
    }
}).start());

输出（顺序随机）

[Thread-0]: 0
[Thread-4]: 4
[Thread-1]: 1
[Thread-3]: 3
[Thread-2]: 2

←Back to home