手机网站开发语言选择,h5网站建站,dede网站地图模板,合肥网络开发公司10.Tomcat线程池 LimitLatch 用来限流#xff0c;可以控制最大连接个数#xff0c;类似 J.U.C 中的 Semaphore 后面再讲 Acceptor 只负责【接收新的 socket 连接】 Poller 只负责监听 socket channel 是否有【可读的 I/O 事件】 一旦可读#xff0c;封装一个任务对象#x…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 RecursiveTaskInteger{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
15J.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 locknew 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 syncnew 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-1state 1 head 指向刚刚 Thread-1 所在的 Node该 Node 清空 Thread 原本的 head 因为从链表断开而可被垃圾回收 如果这时候有其它线程来竞争非公平的体现例如这时有 Thread-4 来了
如果不巧又被 Thread-4 占了先 Thread-4 被设置为 exclusiveOwnerThreadstate 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()) {
// stateint 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 状态为 -2Node.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 改为 0Thread-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 dataContainernew 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 rrw.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.lockt2 r.lock
1 t1 成功上锁流程与 ReentrantLock 加锁相比没有特殊之处不同是写锁状态占了 state 的低 16 位而读锁使用的是 state 的高 16 位 2t2 执行 r.lock这时进入读锁的 sync.acquireShared(1) 流程首先会进入 tryAcquireShared 流程。如果有写锁占据那么 tryAcquireShared 返回 -1 表示失败
tryAcquireShared 返回值表示 -1 表示失败 0 表示成功但后继节点不会继续唤醒 正数表示成功而且数值是还有几个后继节点需要唤醒读写锁返回 1 3这时会进入 sync.doAcquireShared(1) 流程首先也是调用 addWaiter 添加节点不同之处在于节点被设置为Node.SHARED 模式而非 Node.EXCLUSIVE 模式注意此时 t2 仍处于活跃状态
4t2 会看看自己的节点是不是老二如果是还会再次调用 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.unlockt3 r.unlock t2 进入 sync.releaseShared(1) 中调用 tryReleaseShared(1) 让计数减一但由于计数还不为零 t3 进入 sync.releaseShared(1) 中调用 tryReleaseShared(1) 让计数减一这回计数为零了进入doReleaseShared() 将头节点从 -1 改为 0 并唤醒老二即 之后 t4 在 acquireQueued 中 parkAndCheckInterrupt 处恢复运行再次 for (;;) 这次自己是老二并且没有其他竞争tryAcquire(1) 成功修改头结点流程结束
