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) 成功,修改头结点,流程结束