多线程学习Day09

10.Tomcat线程池

LimitLatch 用来限流，可以控制最大连接个数，类似 J.U.C 中的 Semaphore 后面再讲
Acceptor 只负责【接收新的 socket 连接】
Poller 只负责监听 socket channel 是否有【可读的 I/O 事件】
一旦可读，封装一个任务对象（socketProcessor），提交给 Executor 线程池处理
Executor 线程池中的工作线程最终负责【处理请求】 

Tomcat 线程池扩展了 ThreadPoolExecutor，行为稍有不同
       如果总线程数达到 maximumPoolSize
       这时不会立刻抛 RejectedExecutionException 异常
而是再次尝试将任务放入队列，如果还失败，才抛出 RejectedExecutionException 异常 

 public void execute(Runnable command, long timeout, TimeUnit unit) {submittedCount.incrementAndGet();try {super.execute(command);} catch (RejectedExecutionException rx) {if (super.getQueue() instanceof TaskQueue) {final TaskQueue queue = (TaskQueue)super.getQueue();try {if (!queue.force(command, timeout, unit)) {submittedCount.decrementAndGet();throw new RejectedExecutionException("Queue capacity is full.");}} catch (InterruptedException x) {submittedCount.decrementAndGet();Thread.interrupted();throw new RejectedExecutionException(x);}} else {submittedCount.decrementAndGet();throw rx;}}}

Connector 配置

 Executor 线程配置

Fork/Join

 Fork/Join 是 JDK 1.7 加入的新的线程池实现，它体现的是一种分治思想，适用于能够进行任务拆分的 cpu 密集型运算
所谓的任务拆分，是将一个大任务拆分为算法上相同的小任务，直至不能拆分可以直接求解。跟递归相关的一些计算，如归并排序、斐波那契数列、都可以用分治思想进行求解
Fork/Join 在分治的基础上加入了多线程，可以把每个任务的分解和合并交给不同的线程来完成，进一步提升了运算效率
Fork/Join 默认会创建与 cpu 核心数大小相同的线程池

提交给 Fork/Join 线程池的任务需要继承 RecursiveTask（有返回值）或 RecursiveAction（没有返回值），例如下面定义了一个对 1~n 之间的整数求和的任务

@Slf4j(topic = "c.TestForkJoin")
public class TestForkJoin {public static void main(String[] args) {ForkJoinPool pool =new ForkJoinPool(4);System.out.println(pool.invoke(new MyTask(5)));}
}
@Slf4j(topic = "c.MyTask")
class MyTask extends RecursiveTask<Integer>{private int n;public MyTask(int n) {this.n = n;}@Overridepublic String toString() {return "{" + n + '}';}@Overrideprotected Integer compute() {
// 如果 n 已经为 1，可以求得结果了if (n == 1) {log.debug("join() {}", n);return n;}
// 将任务进行拆分(fork)MyTask t1 = new MyTask(n - 1);t1.fork();log.debug("fork() {} + {}", n, t1);
// 合并(join)结果int result = n + t1.join();log.debug("join() {} + {} = {}", n, t1, result);return result;}
}

结果

09:54:35 [ForkJoinPool-1-worker-7] c.MyTask - fork() 3 + {2}
09:54:35 [ForkJoinPool-1-worker-7] c.MyTask - join() 1
09:54:35 [ForkJoinPool-1-worker-1] c.MyTask - fork() 2 + {1}
09:54:35 [ForkJoinPool-1-worker-3] c.MyTask - fork() 5 + {4}
09:54:35 [ForkJoinPool-1-worker-5] c.MyTask - fork() 4 + {3}
09:54:35 [ForkJoinPool-1-worker-1] c.MyTask - join() 2 + {1} = 3
09:54:35 [ForkJoinPool-1-worker-7] c.MyTask - join() 3 + {2} = 6
09:54:35 [ForkJoinPool-1-worker-5] c.MyTask - join() 4 + {3} = 10
09:54:35 [ForkJoinPool-1-worker-3] c.MyTask - join() 5 + {4} = 15
15

J.U.C

 AQS原理

概述

全称是 AbstractQueuedSynchronizer，是阻塞式锁和相关的同步器工具的框架

特点：
用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁

                 getState - 获取 state 状态
                 setState - 设置 state 状态
                 compareAndSetState - cas 机制设置 state 状态
                 独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源
提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList
条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet

子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）
     tryAcquire
     tryRelease
     tryAcquireShared
     tryReleaseShared
     isHeldExclusively

获取锁的姿势

// 如果获取锁失败
if (!tryAcquire(arg)) {
// 入队, 可以选择阻塞当前线程 park unpark
}

释放锁的姿势

// 如果释放锁成功
if (tryRelease(arg)) {
// 让阻塞线程恢复运行
}

 一个自定义锁

@Slf4j(topic = "c.TestAQS")
public class TestAQS {public static void main(String[] args) {Mylock lock=new Mylock();new Thread(()->{lock.lock();try {log.debug("locking..");Thread.sleep(1000);} catch (InterruptedException e) {throw new RuntimeException(e);} finally {log.debug("unlocking...");lock.unlock();}},"t1").start();new Thread(()->{lock.lock();try {log.debug("locking..");} finally {log.debug("unlocking...");lock.unlock();}},"t2").start();}
}
//自定义锁，不可重入
class Mylock implements Lock{//独占锁class MySync extends AbstractQueuedSynchronizer{@Overrideprotected boolean tryAcquire(int arg) {if(compareAndSetState(0,1)){//加上了锁,并设置持有者为当前线程setExclusiveOwnerThread(Thread.currentThread());return true;}return false;}@Overrideprotected boolean tryRelease(int arg) {setExclusiveOwnerThread(null);setState(0);return true;}@Override//是否持有独占锁protected boolean isHeldExclusively() {return getState()==1;}public Condition newCondition(){return new ConditionObject();}}private MySync sync=new MySync();@Override//加锁，不成功进入等待队列public void lock() {sync.acquire(1);}@Override//加锁，可打断public void lockInterruptibly() throws InterruptedException {sync.acquireInterruptibly(1);}@Override//尝试加锁，尝试一次public boolean tryLock() {return sync.tryAcquire(1);}@Override//尝试加锁，带超时public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {return sync.tryAcquireNanos(1,unit.toNanos(time));}@Override//解锁public void unlock() {sync.release(1);}@Overridepublic Condition newCondition() {return sync.newCondition();}
}

10:21:46 [t1] c.TestAQS - locking..
10:21:47 [t1] c.TestAQS - unlocking...
10:21:47 [t2] c.TestAQS - locking..
10:21:47 [t2] c.TestAQS - unlocking...

ReentrantLock 原理

1.非公平锁实现原理 

构造器默认为非公平锁实现

public ReentrantLock() {
sync = new NonfairSync();
}

NonfairSync 继承自 AQS
没有竞争时

第一个竞争出现时

 

Thread-1 执行了
1. CAS 尝试将 state 由 0 改为 1，结果失败
2. 进入 tryAcquire 逻辑，这时 state 已经是1，结果仍然失败
3. 接下来进入 addWaiter 逻辑，构造 Node 队列
      图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态
      Node 的创建是懒惰的
      其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程 

当前线程进入 acquireQueued 逻辑
1. acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
2. 如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
3. 进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false

 4. shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时state 仍为 1，失败
5. 当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回true
6. 进入 parkAndCheckInterrupt， Thread-1 park（灰色表示）

再次有多个线程经历上述过程竞争失败，变成这个样子

Thread-0 释放锁，进入 tryRelease 流程，如果成功
设置 exclusiveOwnerThread 为 nullstate = 0

当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程
找到队列中离 head 最近的一个 Node（没取消的），unpark 恢复其运行，本例中即为 Thread-1
回到 Thread-1 的 acquireQueued 流程

如果加锁成功（没有竞争），会设置
     exclusiveOwnerThread 为 Thread-1，state = 1
     head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread
     原本的 head 因为从链表断开，而可被垃圾回收
如果这时候有其它线程来竞争（非公平的体现），例如这时有 Thread-4 来了

如果不巧又被 Thread-4 占了先
Thread-4 被设置为 exclusiveOwnerThread，state = 1
Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞 

2.可重入原理

static final class NonfairSync extends Sync {// ...
// Sync 继承过来的方法, 方便阅读, 放在此处final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {if (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}
// 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入else if (current == getExclusiveOwnerThread()) {
// state++int nextc = c + acquires;if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}// Sync 继承过来的方法, 方便阅读, 放在此处protected final boolean tryRelease(int releases) {
// state--int c = getState() - releases;if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();boolean free = false;
// 支持锁重入, 只有 state 减为 0, 才释放成功if (c == 0) {free = true;setExclusiveOwnerThread(null);}setState(c);return free;}
}

3.可打断原理

不可打断模式

在此模式下，即使它被打断，仍会驻留在 AQS 队列中，一直要等到获得锁后方能得知自己被打断了

// Sync 继承自 AQS
static final class NonfairSync extends Sync {// ...private final boolean parkAndCheckInterrupt() {
// 如果打断标记已经是 true, 则 park 会失效LockSupport.park(this);
// interrupted 会清除打断标记return Thread.interrupted();}final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {boolean interrupted = false;for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null;failed = false;
// 还是需要获得锁后, 才能返回打断状态return interrupted;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) {
// 如果是因为 interrupt 被唤醒, 返回打断状态为 trueinterrupted = true;}}} finally {if (failed)cancelAcquire(node);}}public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {
// 如果打断状态为 trueselfInterrupt();}}static void selfInterrupt() {
// 重新产生一次中断Thread.currentThread().interrupt();}
}

可打断模式

static final class NonfairSync extends Sync {public final void acquireInterruptibly(int arg) throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();
// 如果没有获得到锁, 进入 ㈠if (!tryAcquire(arg))doAcquireInterruptibly(arg);}// ㈠ 可打断的获取锁流程private void doAcquireInterruptibly(int arg) throws InterruptedException {final Node node = addWaiter(Node.EXCLUSIVE);boolean failed = true;try {for (;;) {final Node p = node.predecessor();if (p == head && tryAcquire(arg)) {setHead(node);p.next = null; // help GCfailed = false;return;}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt()) {
// 在 park 过程中如果被 interrupt 会进入此
// 这时候抛出异常, 而不会再次进入 for (;;)throw new InterruptedException();}}} finally {if (failed)cancelAcquire(node);}}
}

4.公平锁实现原理

与非公平锁主要区别在于 tryAcquire 方法的实现

static final class FairSync extends Sync {private static final long serialVersionUID = -3000897897090466540L;final void lock() {acquire(1);}// AQS 继承过来的方法, 方便阅读, 放在此处public final void acquire(int arg) {if (!tryAcquire(arg) &&acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) {selfInterrupt();}}// 与非公平锁主要区别在于 tryAcquire 方法的实现protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {
// 先检查 AQS 队列中是否有前驱节点, 没有才去竞争if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}else if (current == getExclusiveOwnerThread()) {int nextc = c + acquires;if (nextc < 0)throw new Error("Maximum lock count exceeded");setState(nextc);return true;}return false;}// ㈠ AQS 继承过来的方法, 方便阅读, 放在此处public final boolean hasQueuedPredecessors() {Node t = tail;Node h = head;Node s;
// h != t 时表示队列中有 Nodereturn h != t &&(
// (s = h.next) == null 表示队列中还有没有老二(s = h.next) == null ||// 或者队列中老二线程不是此线程s.thread != Thread.currentThread());}
}

5.条件变量实现原理

每个条件变量其实就对应着一个等待队列，其实现类是 ConditionObject

await 流程

开始 Thread-0 持有锁，调用 await，进入 ConditionObject 的 addConditionWaiter 流程
创建新的 Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部

接下来进入 AQS 的 fullyRelease 流程，释放同步器上的锁

unpark AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 Thread-1 竞争成功

park 阻塞 Thread-0

signal 流程

假设 Thread-1 要来唤醒 Thread-0

进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node

执行 transferForSignal 流程，将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的waitStatus 改为 -1

读写锁

ReentrantReadWriteLock

当读操作远远高于写操作时，这时候使用 读写锁 让 读-读 可以并发，提高性能。 类似于数据库中的 select ...from ... lock in share mode
提供一个 数据容器类内部分别使用读锁保护数据的 read() 方法，写锁保护数据的 write() 方法

@Slf4j(topic = "c.TestReadWriteLock")
public class TestReadWriteLock {public static void main(String[] args) {DataContainer dataContainer=new DataContainer();new Thread(()->{dataContainer.read();},"t1").start();new Thread(()->{dataContainer.read();},"t2").start();}
}
@Slf4j(topic = "c.DataContainer")
class DataContainer{private Object data;private ReentrantReadWriteLock rw =new ReentrantReadWriteLock();private ReentrantReadWriteLock.ReadLock r=rw.readLock();private ReentrantReadWriteLock.WriteLock w= rw.writeLock();public Object read(){log.debug("获取读锁");r.lock();try {log.debug("读取");Thread.sleep(1000);return data;} catch (InterruptedException e) {throw new RuntimeException(e);} finally {log.debug("释放读锁");r.unlock();}}public void write(){log.debug("获取写锁");w.lock();try {log.debug("写入");}finally {log.debug("释放写锁");w.unlock();}}
}

这里可以看到虽然加了锁但是读和读之间时不互斥的，那这里我有个疑问，这个锁的意义是什么呢，尽管多个读操作可以同时进行，但是一旦有线程尝试写操作，写锁将确保所有读操作已经完成，即在写锁被获取之前不允许新的读锁被获取。这确保了在写入新数据时，没有读操作会看到不一致的状态。也就是说除了读读其他想读写和写写都是互斥的。

15:49:00 [t1] c.DataContainer - 获取读锁
15:49:00 [t1] c.DataContainer - 读取
15:49:00 [t2] c.DataContainer - 获取读锁
15:49:00 [t2] c.DataContainer - 读取
15:49:01 [t1] c.DataContainer - 释放读锁
15:49:01 [t2] c.DataContainer - 释放读锁

读锁不支持条件变量
重入时升级不支持：即持有读锁的情况下去获取写锁，会导致获取写锁永久等待，但是它支持降级

缓存 

缓存更新策略

先清缓存

先更新数据库

 补充一种情况，假设查询线程 A 查询数据时恰好缓存数据由于时间到期失效，或是第一次查询

读写锁原理

读写锁用的是同一个 Sycn 同步器，因此等待队列、state 等也是同一个

t1 w.lock，t2 r.lock

1） t1 成功上锁，流程与 ReentrantLock 加锁相比没有特殊之处，不同是写锁状态占了 state 的低 16 位，而读锁使用的是 state 的高 16 位

2）t2 执行 r.lock，这时进入读锁的 sync.acquireShared(1) 流程，首先会进入 tryAcquireShared 流程。如果有写锁占据，那么 tryAcquireShared 返回 -1 表示失败

tryAcquireShared 返回值表示  -1 表示失败   0 表示成功，但后继节点不会继续唤醒  正数表示成功，而且数值是还有几个后继节点需要唤醒，读写锁返回 1

3）这时会进入 sync.doAcquireShared(1) 流程，首先也是调用 addWaiter 添加节点，不同之处在于节点被设置为Node.SHARED 模式而非 Node.EXCLUSIVE 模式，注意此时 t2 仍处于活跃状态

4）t2 会看看自己的节点是不是老二，如果是，还会再次调用 tryAcquireShared(1) 来尝试获取锁

5）如果没有成功，在 doAcquireShared 内 for (;;) 循环一次，把前驱节点的 waitStatus 改为 -1，再 for (;;) 循环一次尝试 tryAcquireShared(1) 如果还不成功，那么在 parkAndCheckInterrupt() 处 park

这种状态下，假设又有 t3 加读锁和 t4 加写锁，这期间 t1 仍然持有锁，就变成了下面的样子

这时会走到写锁的 sync.release(1) 流程，调用 sync.tryRelease(1) 成功，变成下面的样子

接下来执行唤醒流程 sync.unparkSuccessor，即让老二恢复运行，这时 t2 在 doAcquireShared 内
parkAndCheckInterrupt() 处恢复运行，这回再来一次 for (;;) 执行 tryAcquireShared 成功则让读锁计数加一

这时 t2 已经恢复运行，接下来 t2 调用 setHeadAndPropagate(node, 1)，它原本所在节点被置头节点

事情还没完，在 setHeadAndPropagate 方法内还会检查下一个节点是否是 shared，如果是则调用
doReleaseShared() 将 head 的状态从 -1 改为 0 并唤醒老二，这时 t3 在 doAcquireShared 内
parkAndCheckInterrupt() 处恢复运行

这回再来一次 for (;;) 执行 tryAcquireShared 成功则让读锁计数加一

这时 t3 已经恢复运行，接下来 t3 调用 setHeadAndPropagate(node, 1)，它原本所在节点被置为头节点

下一个节点不是 shared 了，因此不会继续唤醒 t4 所在节点

t2 r.unlock，t3 r.unlock
t2 进入 sync.releaseShared(1) 中，调用 tryReleaseShared(1) 让计数减一，但由于计数还不为零

t3 进入 sync.releaseShared(1) 中，调用 tryReleaseShared(1) 让计数减一，这回计数为零了，进入doReleaseShared() 将头节点从 -1 改为 0 并唤醒老二，即

之后 t4 在 acquireQueued 中 parkAndCheckInterrupt 处恢复运行，再次 for (;;) 这次自己是老二，并且没有其他竞争，tryAcquire(1) 成功，修改头结点，流程结束