java.util.concurrent下相关研究学习

编程入门行业动态更新时间:2024-10-28 20:18:11

写在前面

这个包对象是JDK 1.5,1.6 ,1.8 之后，新增的，大概2009年，新增了，枚举定义，基本类型包装，注解，加强for循环，等新特性，提高了java对并发编程的支持，也扩大了Java的市场，前景

JDK 1.8 下 JUC 下共有 17（atomic包下） + 10（lock包下） + 58 个对象组成；
在后续的Java 高版本中，有更改 17（atomic包下） + 10（lock包下） + 61个对象组成，多了的 3 个对象是

java.util.concurrent.Flowjava.util.concurrent.Helpersjava.util.concurrent.SubmissionPublisher

一、lock 包下

这个包下共 10 个对象（接口、类）

如图，

下面简单介绍，这10 个对象

1.1、AbstractOwnableSynchronizer

可参考文章，介绍了AbstractOwnableSynchronizer和AbstractQueuedSynchronizer，链接

1.2、AbstractQueuedLongSynchronizer

1.3、AbstractQueuedSynchronizer

这个就是简称 AQS 的东西，
java.util.concurrent包中很多类都依赖于这个类所提供队列式同步器，比如说常用的ReentranLock，Semaphore和CountDownLatch

代码示例 RoboVM(java 创建 IOS APP框架)

/*** Acquires in shared mode, aborting if interrupted.  Implemented* by first checking interrupt status, then invoking at least once* {@link #tryAcquireShared}, returning on suess.  Otherwise the* thread is queued, possibly repeatedly blocking and unblocking,* invoking {@link #tryAcquireShared} until suess or the thread* is interrupted.* @param arg the acquire argument.* This value is conveyed to {@link #tryAcquireShared} but is* otherwise uninterpreted and can represent anything* you like.* @throws InterruptedException if the current thread is interrupted*/public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();if (tryAcquireShared(arg) < 0)doAcquireSharedInterruptibly(arg);}

1.4、Condition

在使用Lock之前，我们使用的最多的同步方式应该是synchronized关键字来实现同步方式了。配合Object的wait()、notify()系列方法可以实现等待/通知模式。Condition接口也提供了类似Object的监视器方法，与Lock配合可以实现等待/通知模式

简介，可参考，链接

1.4.1、apache-druid中使用示例

  private static final TimeUnit TIME_UNIT = TimeUnit.MILLISECONDS;@VisibleForTestingfinal ArrayDeque<T> objects;private final ReentrantLock lock;private final Condition notEnough;private final int maxSize;......@Nullableprivate T pollObject(long timeoutMs) throws InterruptedException{long nanos = TIME_UNIT.toNanos(timeoutMs);final ReentrantLock lock = this.lock;lock.lockInterruptibly();try {while (objects.isEmpty()) {if (nanos <= 0) {return null;}nanos = notEnough.awaitNanos(nanos);}return objects.pop();}finally {lock.unlock();}}

1.5、Lock

Jdk中核心实现类包括一下

代码示例 apache/hive 任务调度中

private final Lock scheduleLock = new ReentrantLock()
....private void trySchedulingPendingTasks() {scheduleLock.lock();try {pendingScheduleInvocations.set(true);scheduleCondition.signal();} finally {scheduleLock.unlock();}}

1.6、LockSupport

代码示例 h2oai/h2o-2

public boolean block() {if (isReleasable())return true;else if (!timed)LockSupport.park(this);else if (nanos > 0)LockSupport.parkNanos(this, nanos);return isReleasable();}}

1.7、ReadWriteLock

Synchronized存在明显的一个性能问题就是读与读之间互斥，
ReadWriteLock管理一组锁，一个是只读的锁，一个是写锁。可以做到读和读互不影响，读和写互斥，写和写互斥，提高读写的效率
Java并发库中ReetrantReadWriteLock实现了ReadWriteLock接口并添加了可重入的特性。

代码示例 apache/rocketMQ

private ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
...
...
public String getAllConfigsFormatString() {try {readWriteLock.readLock().lockInterruptibly();try {return getAllConfigsInternal();} finally {readWriteLock.readLock().unlock();}} catch (InterruptedException e) {log.error("getAllConfigsFormatString lock error");}return null;}

代码示例 jenkinsci/jenkins

 /*protected*/ void start(WorkUnit task) {lock.writeLock().lock();try {this.workUnit = task;super.start();started = true;} finally {lock.writeLock().unlock();}}

1.8、ReentrantLock

java除了使用关键字synchronized外，还可以使用ReentrantLock实现独占锁的功能。而且ReentrantLock相比synchronized而言功能更加丰富，使用起来更为灵活，也更适合复杂的并发场景

构造参数为 true代表公平锁，会根据入排队时间，优先获取锁

@Overridepublic Lock get() {return new ReentrantLock(false);}});

代码示例 alibaba/druid

public void setEnable(boolean enable) {lock.lock();try {this.enable = enable;if (!enable) {notEmpty.signalAll();notEmptySignalCount++;}} finally {lock.unlock();}}

1.9、ReentrantReadWriteLock

代码示例 apache / hive

 private final ConcurrentMap<QueryIdentifier, ReadWriteLock> dagSpecificLocks = new ConcurrentHashMap<>();......private ReentrantReadWriteLock getDagLock(QueryIdentifier queryIdentifier) {lock.lock();try {ReentrantReadWriteLock dagLock = dagSpecificLocks.get(queryIdentifier);if (dagLock == null) {dagLock = new ReentrantReadWriteLock();dagSpecificLocks.put(queryIdentifier, dagLock);}return dagLock;} finally {lock.unlock();}}

代码示例 oracle/opengrok

  private RuntimeEnvironment() {configuration = new Configuration();configLock = new CloseableReentrantReadWriteLock();watchDog = new WatchDogService();lzIndexerParallelizer = LazilyInstantiate.using(() ->new IndexerParallelizer(this));lzSearchExecutor = LazilyInstantiate.using(() -> newSearchExecutor());lzRevisionExecutor = LazilyInstantiate.using(() -> newRevisionExecutor());}....../*** Add repositories to the list.* @param repositories list of repositories*/public void addRepositories(List<RepositoryInfo> repositories) {Lock writeLock = configLock.writeLock();try {writeLock.lock();configuration.addRepositories(repositories);} finally {writeLock.unlock();}}

1.10、StampedLock

这个对象是 JDK 1.8 之后出现的对象，作为读写锁

StampedLock和ReadWriteLock相比，改进之处在于：读的过程中也允许获取写锁后写入！这样一来，我们读的数据就可能不一致，所以，需要一点额外的代码来判断读的过程中是否有写入，这种读锁是一种乐观锁。

乐观锁的意思就是乐观地估计读的过程中大概率不会有写入，因此被称为乐观锁。反过来，悲观锁则是读的过程中拒绝有写入，也就是写入必须等待。显然乐观锁的并发效率更高，但一旦有小概率的写入导致读取的数据不一致，需要能检测出来，再读一遍就行。

代码示例 apache/pulsar

private final StampedLock rwLock = new StampedLock();
...
private final ArrayList<Item> heap = Lists.newArrayList();
...public boolean isEmpty() {long stamp = rwLock.tryOptimisticRead();boolean isEmpty = heap.isEmpty();if (!rwLock.validate(stamp)) {// Fallback to read lockstamp = rwLock.readLock();try {isEmpty = heap.isEmpty();} finally {rwLock.unlockRead(stamp);}}return isEmpty;}