ConcurrentHashMap

1.7的ConcurrentHashMap：分而治之--segment分段锁

segment继承自Reentrantlock，实现加锁

根据hash分配到不同的segment中，这样当我操作这个concurrenthashmap时，只会锁住一段我需要的部分，其他未使用到的部分仍然可以被其他线程访问，保证了并发下的效率。

• concurrencyLevel:并发数，默认16,也就是16个segment数组

• segment的数量创建后不可更改，但是每个segment内部元素可以扩容：

  • 初始大小16

  • 负载因子0.75

• 初始化槽时通过cas保证只有一个线程初始化成功

• put时若获取独占锁失败，快速scanandlockforput获取锁

• rehash扩容，扩容单个segment内元素（kv）为原来的2倍，无需加锁，此时已是独占锁状态

• get：进行hash，根据hash值找到对应的segment槽，再去segment里找数组具体位置，顺便遍历数组里的链表。

tabAt castabAt

static final <K,V&gt; Node<K,V&gt; tabAt(Node<K,V&gt;[] tab, int i) {
   return (Node<K,V&gt;)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
​
static final <K,V&gt; boolean casTabAt(Node<K,V&gt;[] tab, int i,
                                   Node<K,V&gt; c, Node<K,V&gt; v) {
   return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}

1.8的ConcurrentHashMap：CAS volatile synchronized配合实现，put get 包括init，扩容前都经过cas处理

• 数组小于64时不会转成红黑树

• 数组达到负载因子0.75阈值后扩容为原来的2倍

• transfor方法实现数据迁移

• put ，分支初始化，判断是数组还是红黑树，并做相应的put

CopyOnwriteArrayList

• 内部有Reentrantlock，在每次操作前上锁保证并发安全

• 内部类的迭代器：COWIterator，默认的add set remove操作都抛出UnsupportedOperationsException

  • fail-safe：安全失败，在迭代时保存了一份原对象引用的快照，不会抛出ConcurrentModificationException

  • fail-fast：快速失败：防止并发环境下对同一个对象非法操作，在迭代时提前抛出异常。

• 在add set remove 操作时都会上锁，并通过copyof操作复制一份副本

• 底层的copyOf也是和ArrayList一样调用的Arrays.copyOf()，内部是一个native的arraycopyof方法。

  public E set(int index, E element) {
     final ReentrantLock lock = this.lock;
     lock.lock();
     try {
         Object[] elements = getArray();
         E oldValue = get(elements, index);
  ​
         if (oldValue != element) {
             int len = elements.length;
             Object[] newElements = Arrays.copyOf(elements, len);
             newElements[index] = element;
             setArray(newElements);
        } else {
             // Not quite a no-op; ensures volatile write semantics
             setArray(elements);
        }
         return oldValue;
    } finally {
         lock.unlock();
    }
  }

  缺点：

  1. 写时复制保证并发安全，意味着每次set add remove都会复制一份副本，非常的耗费内存，可能频繁从触发gc

  2. 实时性不高，拷贝，新增元素都需要时间，只能保证最终一致性

  3. 虽然适合读多写少的场景，但是线上数据要是过多也容易崩

ConcurrentLinkedQueue

• 继承AbstractQueue，实现Queue接口

• 无锁cas实现的并发队列，有很高的并发性能，但是使用场景没有阻塞队列常用，需要高并发无锁队列场景可用

• 队列的常用方法：offer poll，remove

• pffer poll时都有hops延迟更新

BlockingQueue

• 阻塞队列

• 常用于生产者消费者场景

• 生产者从队尾入队，消费者从队头获取

• 队列为空时，消费者阻塞，等待生产者投递生产

• 队列满时，生产者阻塞，等待消费者消费

提供的方法：

如果无法执行相应：	抛出异常	成功true 失败false	阻塞	超时
插入	add(0)	offer(0)	put(0)	offer(o,timeout,timeunit)
移除	remove()	poll()	take()	pool(o,timeout,timeunit)
检查	element()	peek()

BlockingDeque

双向队列，实现了BlockingQueue接口，可从队列两头存取

自然有相对应的addFirst,addLast,pollFrist,pollLast。。。。

ArrayBlockingQueue

有界的阻塞队列，实现了BlockingQueue接口，基于数组实现，有数组的特性，初始化后大小固定不能更改

DelayQueue

延迟特性的队列，实现了BlockingQueue接口，其内部元素需要实现Delayed接口

public class DelayQueue<E extends Delayed&gt; extends AbstractQueue<E&gt;
   implements BlockingQueue<E&gt; {

通过getDelayed()返回对象，在一定延迟过后才会返回对象。-1为立即返回

LinkedBlockingQueue

无界的阻塞队列，底层链表实现，不设置队列上限则默认为int的最大值Integer.MAX_VALUE（2147483647），元素过多一直阻塞，导致内存泄漏(?)，jdk8还有remove导致死循环bug，jdk9修复。

public LinkedBlockingQueue() {
   this(Integer.MAX_VALUE);
}

PriorityBlockingQueue

带有优先级的阻塞队列，无法插入null值，插入的元素必需实现Comparable接口自己实现排序

• 自带的Iterator遍历时不保证按优先顺序遍历

SynchronousQueue

同步队列，内部同时只能容纳单个队列，叫做同步点更合适。

• 队列没对象，取的人取的时候阻塞

• 队列有对象，放的人放的时候阻塞

工具类

异步编排：异步线程之间对任务的先后顺序性有需求，可使用juc相关的工具类来实现线程之间的调度。（我理解为对于Object的wait notify线程调度的功能，juc下提供更多封装好了，功能更全的拓展工具类实现。（内部类Sync，基于AQS等待队列自旋实现线程调度功能）

下面只汇总常用工具类以及介绍大体功能场景，不再赘述深入源码层面。

CountDownLatch

倒计时，提供n，异步的线程完成任务时调用countdown，计数器-1，当计数器为0时代表全部任务完成。

• 影响的是主线程

• 一次性使用

• 内部类Sync继承自AQS

• await方法调用AQS的acquireSharedInterruptibly，AQS调用Sync实现的tryAcqurieShared（CAS获取公平锁）成功后进入

  doAcquireSharedInterruptibly，自旋进入等待队列等待释放（或中断）

• lol进游戏，所有玩家加载完成后才能开始对局

示例：其他线程countdown只到count为0时，await方才会被唤醒

package com.kaguya.demo;
​
import java.util.concurrent.CountDownLatch;
​
public class JUCUtils {
   public static void main(String[] args) throws InterruptedException {
       // 异步执行两个业务，都执行完毕后主方法再进行总处理
       CountDownLatch countDownLatch = new CountDownLatch(2);
       Thread t1 = new Thread(() -> {
           try {
               Thread.sleep(500);
          } catch (InterruptedException e) {
               e.printStackTrace();
          }
           System.out.println("执行业务1，查询分类导航数据..完毕");
           countDownLatch.countDown();
          },"t1");
​
       Thread t2 = new Thread(() -> {
           try {
               Thread.sleep(500);
          } catch (InterruptedException e) {
               e.printStackTrace();
          }
           System.out.println("执行业务2，查询商品分页列表..完毕");
           countDownLatch.countDown();
      },"t2");
       long begin = System.currentTimeMillis();
       t2.start();
       t1.start();
​
       countDownLatch.await();
       System.out.println("所有异步业务执行完毕，耗时。。"+ (System.currentTimeMillis() - begin));
  }
}

CyclicBarrier

循环栅栏，设定n个栅栏数，当n个线程都执行到栅栏处时才会唤醒其他线程，当全部执行到屏障处时可执行唤醒回调

• 不影响主线程

• 可以复用

• 计数器累加

• 基于ReentrantLock+Condition（内部通过AQS等待队列实现，doawait判断中断和index，不满足条件自旋等待，满足后调用signal唤醒，有屏障回调就执行）

• 网游玩家开始对局，电脑配置好的人先进游戏，等待全部玩家进入后才正式开局

示例：

public static void cyclicBarrier() {
   // 6个线程批量导入，全部导入成功后保存一条日志数据记录耗时
   long begin = System.currentTimeMillis();
   CyclicBarrier barrier = new CyclicBarrier(6, () -> {
       System.out.println("批量插入成功，总耗时：" + (System.currentTimeMillis() - begin));
  });
   for (int i = 0; i < 6; i++) {
       new Thread(() -> {
           try {
               Thread.sleep(500);
               System.out.println("执行批量新增。。。完毕");
               barrier.await();
               System.out.println("确认所有批量新增执行完毕，回头去做自己的事情。（回收资源）");
          } catch (InterruptedException e) {
               e.printStackTrace();
          } catch (BrokenBarrierException e) {
               e.printStackTrace();
          }
      }).start();
  }
}

Semaphore

计数信号量，n多个任务可同时访问资源，分发了信号量的任务才可使用资源。

场景：限流，线程池控制最大连接数，第三方接口有调用数量限制等

• 基于AQS，内部类Sync，子类FairSync noFairSync，故有非公平锁和公平锁实现

• 可以看做是可控参数的CyclicBarrier

• 提供带参数的tryAcquire,acquire,tryRelease,release方法，可释放指定的信号量。

• 公平性由构造参数 new Semaphore(permits, fair) 决定，acquire/tryAcquire 均受此控制

• 默认无参的acquire(),release()方法，只对信号量+1 或 -1

示例：

private static void semaphore() {
   Semaphore semaphore = new Semaphore(10);
   // 饭店10个坐，中午有20个人吃饭，其他人等待别人吃完拍拍肚子走人才能上座
   // 有的人相约两个人一起来吃饭
   for (int i = 0; i < 15; i++) {
       new Thread(() -> {
           try {
               semaphore.acquire();
               System.out.println(Thread.currentThread()+ "上座吃饭。。。");
          } catch (InterruptedException e) {
               e.printStackTrace();
          } finally {
               semaphore.release();
               System.out.println(Thread.currentThread()+ "吃完拍拍肚子走人");
          }
      }).start();
  }
   for (int j = 0; j < 5; j++) {
       new Thread(() -> {
           try {
               semaphore.acquire(2);
               System.out.println(Thread.currentThread()+ "上座吃饭。。。");
          } catch (InterruptedException e) {
               e.printStackTrace();
          } finally {
               System.out.println(Thread.currentThread()+ "吃完拍拍肚子走人");
               semaphore.release(2);
          }
      }).start();
  }
}

ExChange

两个线程之间相互交换数据，A调用exchange后，阻塞，等待b也调用exchange才能唤醒

• 双向的SynchronizedQueue

• 用于两个线程相互通信

• 内部用node[] 解决多线程间竞争问题（没咋深入了解）

示例：

private static void exchange() {
   Exchanger<String&gt; exchanger = new Exchanger<>();
   // 买卖交易，一手交钱，一手交货
   new Thread(() -> {
       System.out.println("卖家准备货物...需要3s");
       try {
           Thread.sleep(3000);
           System.out.println("卖家准备好了货物，等待交换...");
           String res = exchanger.exchange("货物");
           System.out.println("------交易完成，卖家走了，拿到了："+ res);
      } catch (InterruptedException e) {
           e.printStackTrace();
      }
  }).start();
   // 卖家
   new Thread(() -> {
       System.out.println("买家去银行取钱...");
       try {
           Thread.sleep(50);
           System.out.println("买家取完钱了，等待交货...");
           String res = exchanger.exchange("钱");
           System.out.println("------交易完成，买家溜溜，拿到了：" + res);
      } catch (InterruptedException e) {
           e.printStackTrace();
      }
  }).start();
}

Phaser

1.7引入，更高级的CountDownLatch和CyclicBarrier整合增强plus版，可以在运行期间灵活的注册注销信号量，且可有多阶段屏障，可复用

• 适合多线程多阶段提交任务，保证阶段顺序性

• 中途可加入退出

• 考试，需要等所有同学答完题才能进行下一门考试

• 三天的日结兼职，期间经常发生人员变动，每天的日结甲方要求到场人数不同，第一天要10个人，第二天要20个人，第三天要15个人

示例：

ThreadLocal

• 本地线程缓存

• 线程隔离故线程安全

• 本质是ThreadLocalMap，Thread对象中有个ThreadLocalMap的属性，k存当前对象，v存泛型内容

• get()时获取当前线程，并从当前线程拿到它对应的ThreadLocalMap

• 保存当前线程对象的副本，故有内存消耗

• 内部entry是弱引用，但是配合线程池使用会引发内存泄漏，需要手动调用remove方法

• 常用于线程池，session等本地线程唯一不共享的地方