美食网站建设的背景,网站建设费用什么意思,免费网页托管,wordpress打开超级慢池化技术是比较常见的一种技术#xff0c;在平时我们已经就接触很多了#xff0c;比如线程池#xff0c;数据库连接池等等。当我们要使用一个资源的时候从池中去获取#xff0c;用完就放回池中以便其他线程可以使用#xff0c;这样的目的就是为了减少资源开销#xff0c;… 池化技术是比较常见的一种技术在平时我们已经就接触很多了比如线程池数据库连接池等等。当我们要使用一个资源的时候从池中去获取用完就放回池中以便其他线程可以使用这样的目的就是为了减少资源开销提升性能。而Netty作为一个高性能的网络框架在这一块也自然下足了工夫下面我们就来看一下在Netty中的对象池是如何实现的吧 原理 与其他池化实现不同的是其他的池化实现都是全局的但是这样的话在实现的过程中就可能会有并发的问题比如说在获取资源以及回收资源的时候都需要通过加锁等手段去处理而Netty的对象池为了避免这些问题采用了ThreadLocal去为每一个线程创造一个对象池这样的话每一个线程去获取对象以及回收对象的时候就只会在自己所属的对象池中去操作了自然就避免了加锁的过程处理简单来说就是通过空间换时间的思想从而达到了无锁化的目的。 不同的线程池使用独立的对象池虽然解决了上面加锁的问题但是这也会导致另一个问题比如说一个线程从自身的对象池中获取到了一个对象但是这个对象被另外一个线程的对象池拿到并回收了此时该对象就被回收到不属于自己的对象池中了。Netty为了解决这个问题引入了一个队列该队列就是专门存放这些帮助回收的线程回收的对象举个例子线程A从对象池中创建了一个对象这个对象被线程B回收了但是由于这个对象并不属于线程B的所以线程B会为线程A创建一个队列把这个回收对象放到这个队列中当线程A再去从自身对象池中获取对象的时候会先去这个队列中看是否有对象如果有的话就拿出来放回自身线程池中这样就解决了上面的问题了。 所以可以看到Netty对象池中很多都是通过线程隔离的思想去避免线程间并发竞争的情况出现完全体现出了无锁化的设计思想。
源码解析
1整体设计
Recycler
public abstract class RecyclerT {protected abstract T newObject(HandleT handle);
}
Recycler是整个对象池的一个外壳其中提供了一个newObject的抽象方法主要就是给子类去进行实现的子类在可以在该方法中去创建出对象池中的对象
stack
private static final class StackT {// 省略部分代码/*** 存放handle的数组也可以认为是存放对象的数组*/DefaultHandle?[] elements;// 省略部分代码
}
private final FastThreadLocalStackT threadLocal new FastThreadLocalStackT() {Overrideprotected StackT initialValue() {return new StackT(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,interval, maxDelayedQueuesPerThread, delayedQueueInterval);}Overrideprotected void onRemoval(StackT value) {// Let us remove the WeakOrderQueue from the WeakHashMap directly if its safe to remove some overheadif (value.threadRef.get() Thread.currentThread()) {if (DELAYED_RECYCLED.isSet()) {DELAYED_RECYCLED.get().remove(value);}}}
};
stack其实就是真正的对象池实现当每一个线程想去从stack中获取对象的时候都会去从上面的FastThreadLocal中去获取到该线程自己的stack然后再从stack中获取对象。stack中有一个数组该数组就是真是存放对象的但是存放的数据类型是一个DefaultHandle那我们的对象哪里呢其实我们的对象就是在DefaultHandle里面这里DefaultHandle帮我们的对象做了一层包装。
Handle
public interface HandleT extends ObjectPool.HandleT { }public abstract class ObjectPoolT {public interface HandleT {void recycle(T self);}
}
private static final class DefaultHandleT implements HandleT {private static final AtomicIntegerFieldUpdaterDefaultHandle? LAST_RECYCLED_ID_UPDATER;static {AtomicIntegerFieldUpdater? updater AtomicIntegerFieldUpdater.newUpdater(DefaultHandle.class, lastRecycledId);LAST_RECYCLED_ID_UPDATER (AtomicIntegerFieldUpdaterDefaultHandle?) updater;}/*** 当对象被其他帮助回收的线程回收了时候该属性的值就是这个帮助回收的线程的id*/volatile int lastRecycledId;/*** 当原本创建该对象线程从其他帮助回收的线程中拿回该对象的时候此时就会把lastRecycledId的值赋值给recycleId*/int recycleId;boolean hasBeenRecycled;Stack? stack;Object value;DefaultHandle(Stack? stack) {this.stack stack;}Overridepublic void recycle(Object object) {// 校验下回收的对象是否属于当前这个handleif (object ! value) {throw new IllegalArgumentException(object does not belong to handle);}Stack? stack this.stack;if (lastRecycledId ! recycleId || stack null) {throw new IllegalStateException(recycled already);}stack.push(this);}public boolean compareAndSetLastRecycledId(int expectLastRecycledId, int updateLastRecycledId) {// Use weak… because we do not need synchronize-with ordering, only atomicity.// Also, spurious failures are fine, since no code should rely on recycling for correctness.return LAST_RECYCLED_ID_UPDATER.weakCompareAndSet(this, expectLastRecycledId, updateLastRecycledId);}
}
DefaultHandle类实现了Recycler中的Handle接口而Recycler中的Handle接口又继承于ObjectPool的handle接口在ObjectPool的handle接口中有一个recycle方法该方法就是用来回收对象的传入的参数就是要回收的对象当调用recycle方法的时候最终会把当前的handle放到对应的stack中由此可以知道回收对象的入口是在handle而不是stack。另外一个重点的地方就是DefaultHandle的value属性它其实就是我们说所需要的原始对象总体来看stack与handle的关系如下图所示 WeakOrderQueue
当其他线程帮忙回收对象的时候会把对象存放在哪里呢答案就是在WeakOrderQueue中一个帮忙回收的线程针对每一个stack都会有一个WeakOrderQueue去存放回收这个stack的对象并且WeakOrderQueue中通过link指针去构造出一个link链表所以一个WeakOrderQueue也就代表着一个link链表。同理既然WeakOrderQueue是用来存放回收对象的那么这些回收对象也需要被取走是吧所以在一个stack中也构造了一个WeakOrderQueue链表表示当前这个stack被哪些线程的WeakOrderQueue回收了对象当需要取这些回收对象的时候此时就可以取遍历这个WeakOrderQueue即可
Link
一个link对象中包含了一个DefaultHandle数组这个数组存放的就是帮助其他stack回收的对象每一个link之间则形成了链表
小总结
所以综上所述stackhandleWeakOrderQueue与link这4者之间的关系如下图所示引用网上的一张图 最后如果我们要使用Netty的对象池会怎么去使用呢在Netty中提供了模板ObjectPool类去对Recycler进行了包装以便于我们能够更方便地使用ObjectPool类代码如下
public abstract class ObjectPoolT {ObjectPool() { }public abstract T get();public interface HandleT {void recycle(T self);}public interface ObjectCreatorT {T newObject(HandleT handle);}public static T ObjectPoolT newPool(final ObjectCreatorT creator) {return new RecyclerObjectPoolT(ObjectUtil.checkNotNull(creator, creator));}private static final class RecyclerObjectPoolT extends ObjectPoolT {private final RecyclerT recycler;RecyclerObjectPool(final ObjectCreatorT creator) {recycler new RecyclerT() {Overrideprotected T newObject(HandleT handle) {return creator.newObject(handle);}};}Overridepublic T get() {return recycler.get();}}
}
而ObjectPool可以如下使用
ObjectPoolA objectPool ObjectPool.newPool(new ObjectPool.ObjectCreatorA() {Overridepublic A newObject(ObjectPool.HandleA handle) {return new A(handle);}
});A a objectPool.get();
a.recycle();
当然了我们创建的对象需要持有Handle因为回收对象的方法是交给Handle去做的
public class A {private final ObjectPool.HandleA handle;public A(ObjectPool.HandleA handle) {this.handle handle;}public void recycle() {this.handle.recycle(this);}
}
2从stack中获取对象
DefaultHandleT pop() {// 获取到当前线程对应的stack中存在的对象的数量int size this.size;// 条件成立说明stack中没有对象if (size 0) {// 此时需要去从WeakOrderQueue链表中获取对象也就是从其他帮忙回收对象的线程的WeakOrderQueue中获取// 条件成立表示其他帮忙回收对象的线程中也没有对象if (!scavenge()) {// 返回null这样上层会创建新的对象return null;}size this.size;if (size 0) {// double check, avoid racesreturn null;}}// 代码执行到这里说明此时stack中有存活的对象了此时size数量-1size --;// 从数组中获取对象DefaultHandle ret elements[size];elements[size] null;// As we already set the element[size] to null we also need to store the updated size before we do// any validation. Otherwise we may see a null value when later try to pop again without a new element// added before.// 更新size数量this.size size;if (ret.lastRecycledId ! ret.recycleId) {throw new IllegalStateException(recycled multiple times);}// 重置这两个属性ret.recycleId 0;ret.lastRecycledId 0;// 返回对象return ret;
}
3其他线程帮忙回收对象
io.netty.util.Recycler.Stack#pushLater
如果代码来到这里必定就是当前回收对象的线程与创建该对象的线程不是同一个线程
/*** 调用该方法的一定是帮助回收对象的线程。该方法会去把回收的对象放到对应的WeakOrderQueue中* param item 回收的对象* param thread 帮助回收对象的线程也就是当前线程*/
private void pushLater(DefaultHandle? item, Thread thread) {if (maxDelayedQueues 0) {// We dont support recycling across threads and should just drop the item on the floor.return;}// we dont want to have a ref to the queue as the value in our weak map// so we null it out; to ensure there are no races with restoring it later// we impose a memory ordering here (no-op on x86)// 获取这个帮助回收对象的线程对应的WeakOrderQueue// key帮助哪个stack回收// value存放回收这个stack的对象的WeakOrderQueueMapStack?, WeakOrderQueue delayedRecycled DELAYED_RECYCLED.get();WeakOrderQueue queue delayedRecycled.get(this);// 条件成立说明当前线程之前还没有帮助过这个stack回收对象if (queue null) {// 条件成立说明已经超过最大帮助这个stack回收对象的线程上限了if (delayedRecycled.size() maxDelayedQueues) {// 把需要帮助回收对象的stack作为key放到map中对应的value是一个dummy的WeakOrderQueuedelayedRecycled.put(this, WeakOrderQueue.DUMMY);return;}// 代码执行到这里说明此时还没有超过最大帮助这个stack回收对象的线程上限// 给当前stack创建WeakOrderQueue// 条件成立说明当前stack已经没有回收对象的数量去分配了已经不能再创建新的WeakOrderQueue了if ((queue newWeakOrderQueue(thread)) null) {// 放弃回收这个对象return;}// 把stack和对应的WeakOrderQueue放到线程map中delayedRecycled.put(this, queue);}// 在上面的if中当超过最大帮助这个stack回收对象的线程上限的时候就会给map中放入一个dummy的WeakOrderQueueelse if (queue WeakOrderQueue.DUMMY) {// 放弃回收这个对象return;}// 把要回收的对象放到对应的WeakOrderQueue中queue.add(item);
}
/*** 每一个线程都对应一个map* key帮助哪个stack回收对象* value帮助这个stack回收对象时存放对象的WeakOrderQueue*/
private static final FastThreadLocalMapStack?, WeakOrderQueue DELAYED_RECYCLED
new FastThreadLocalMapStack?, WeakOrderQueue() {Overrideprotected MapStack?, WeakOrderQueue initialValue() {return new WeakHashMapStack?, WeakOrderQueue();}
};
首先每个会去从FastThreadLocal中获取到一个mapmap的key表示帮助的是哪一个stack去回收对象value则是已经帮助这个stack回收的对象所存放的队列。
如果该线程是第一次帮助这个stack去回收对象那么就先判断下该线程已经帮助过多少个stack回收对象如果此时已经达到了帮助上限那么此次就不能帮助这个stack回收了然后就给这个stack创建一个DUMMY类型的WeakOrderQueue以便下一次该线程再帮助这个stack回收对象的时候发现对应的队列是一个WeakOrderQueue能够立刻放弃回收如果该线程不是第一次帮助这个stack回收对象了那么能够获取到这个stack对应的WeakOrderQueue队列反之则为这个stack创建一个新的WeakOrderQueue队列并且把这个新创建的WeakOrderQueue队列放到map中最后把要回收的对象放到WeakOrderQueue队列中
创建WeakOrderQueue队列
在上面回收的过程中如果是第一次帮助这个stack回收则需要为其创建一个新的WeakOrderQueue代码如下
private WeakOrderQueue newWeakOrderQueue(Thread thread) {
return WeakOrderQueue.newQueue(this, thread);
}
/*** 给指定的stack创建一个回收对象队列* 该方法可能会被并发调用因为有可能此时会有多个帮助回收对象的线程在调用该方法去为指定的stack创建回收对象队列* param stack 需要帮助回收对象的stack* param thread 帮助回收对象的线程* return WeakOrderQueue对象*/
static WeakOrderQueue newQueue(Stack? stack, Thread thread) {// 条件成立说明指定的stack已经没有回收对象的数量去分配给当前线程去创建回收对象队列了if (!Head.reserveSpaceForLink(stack.availableSharedCapacity)) {// 返回nullreturn null;}// 为stack创建一个新的WeakOrderQueuefinal WeakOrderQueue queue new WeakOrderQueue(stack, thread);// 把这个新创建的WeakOrderQueue放到stack的WeakOrderQueue链表的头部stack.setHead(queue);// 返回这个新创建的WeakOrderQueuereturn queue;
}
可以看到创建WeakOrderQueue之前需要判断对这个stack的availableSharedCapacity属性进行判断那么这个属性是什么意思呢其实就是这个stack有多少个对象是能够被其他线程帮助回收的比方说100个那么其他帮助回收的线程回收这个stack的对象加起来一共最多是100个
static boolean reserveSpaceForLink(AtomicInteger availableSharedCapacity) {for (;;) {// 获取到能够被其他线程回收的对象数量int available availableSharedCapacity.get();// 条件成立说明此时已经没有回收的对象数量了if (available LINK_CAPACITY) {return false;}// 更新可回收的数量对象if (availableSharedCapacity.compareAndSet(available, available - LINK_CAPACITY)) {return true;}}
}
由于可能会有多个帮助这个stack回收对象的线程同时去对availableSharedCapacity进行扣减所以availableSharedCapacity这里通过AtomicInteger修饰保证并发的情况下扣减正确
当创建完WeakOrderQueue之后把这个WeakOrderQueue放到这个stack的WeakOrderQueue链表的头部
/*** 把指定的WeakOrderQueue放到链表的头部头插法。因为有可能会有多个线程同时去给当前的stack创建回收对象线程所以该方法需要加锁执行* 同时这也是整个Netty对象池中唯一加锁的地方* param queue 指定的WeakOrderQueue*/
synchronized void setHead(WeakOrderQueue queue) {queue.setNext(head);head queue;
}
因为上面我们也说了这里可能会有多个线程调用也就是说会有多个线程去创建WeakOrderQueue加到stack的链表头部所以这里需要进行加锁操作需要注意的是这也是整个Netty对象池中唯一一处加锁的地方因为整个Netty对象池的设计就是为了无锁化去设计的
往WeakOrderQueue添加回收对象
/*** 把要回收的handle对象放到当前的WeakOrderQueue的尾link节点中* param handle 要回收的handle对象*/
void add(DefaultHandle? handle) {if (!handle.compareAndSetLastRecycledId(0, id)) {// Separate threads could be racing to add the handle to each their own WeakOrderQueue.// We only add the handle to the queue if we win the race and observe that lastRecycledId is zero.return;}// 这里是为了控制对象回收的频率默认每8次回收才能回收1个对象if (!handle.hasBeenRecycled) {if (handleRecycleCount interval) {handleRecycleCount;// Drop the item to prevent from recycling too aggressively.return;}handleRecycleCount 0;}// 获取到link链表中的尾节点Link tail this.tail;// 当前link的写指针int writeIndex;// 条件成立说明这个link已经放满对象了if ((writeIndex tail.get()) LINK_CAPACITY) {// 创建一个新的link节点Link link head.newLink();// 条件成立说明这个stack中没有可帮助回收的对象数量了if (link null) {// 放弃回收这个对象return;}// 把新创建的link节点设置为尾节点this.tail tail tail.next link;// 重置写指针writeIndex tail.get();}// 把回收对象放到link节点中tail.elements[writeIndex] handle;handle.stack null;// we lazy set to ensure that setting stack to null appears before we unnull it in the owning thread;// this also means we guarantee visibility of an element in the queue if we see the index updated// 更新写指针tail.lazySet(writeIndex 1);
}
WeakOrderQueue中是一个link节点的链表每次添加回收对象的时候都是通过尾插法获取到最后一个link节点的写指针然后根据这个写指针把回收对象放入到这个link节点中。如果link节点已经放满了那么就新创建出一个link节点当然每次创建都要扣减availableSharedCapacity如果扣减完了就返回null此时就放弃回收这个对象了并把这个link节点设置为尾节点最后再把要回收的对象放入到这个新创建的link节点中
4从WeakOrderQueue中获取对象
io.netty.util.Recycler.Stack#scavengeSome
/*** 转移link链表中的对象从头节点开始如果这个link节点中的对象已经转移完了就找下一个link节点进行转移每调一次该方法就转移一个link节点* return 当转移了一个link节点的时候返回true反之返回false*/
private boolean scavengeSome() {// 前一个节点WeakOrderQueue prev;// 当前遍历到的节点WeakOrderQueue cursor this.cursor;// 条件成立说明此时是第一次遍历WeakOrderQueue链表if (cursor null) {prev null;cursor head;// 条件成立说明WeakOrderQueue为空表示没有其他线程帮忙回收对象if (cursor null) {return false;}}// 条件成立不是第一次遍历WeakOrderQueue链表else {prev this.prev;}boolean success false;do {// 条件成立说明转移了回收对象此时跳出循环if (cursor.transfer(this)) {success true;break;}// 代码执行到这里说明当前的WeakOrderQueue中没有转移到回收对象此时获取下一个WeakOrderQueue节点WeakOrderQueue next cursor.getNext();// 条件成立说明这个WeakOrderQueue对象对应的线程挂了也就是帮助回收的线程挂了if (cursor.get() null) {// If the thread associated with the queue is gone, unlink it, after// performing a volatile read to confirm there is no data left to collect.// We never unlink the first queue, as we dont want to synchronize on updating the head.// 把这个WeakOrderQueue中剩余的link节点中的对象进行转移if (cursor.hasFinalData()) {for (;;) {// 每遍历一次就把这个WeakOrderQueue中的一个link节点的回收对象转移一次if (cursor.transfer(this)) {success true;} else {break;}}}if (prev ! null) {// 释放这个WeakOrderQueue所占用的回收对象的数量cursor.reclaimAllSpaceAndUnlink();// 把当前这个WeakOrderQueue从链表中删除prev.setNext(next);}} else {prev cursor;}// 把下一个节点当作当前节点继续遍历cursor next;} while (cursor ! null !success);// 重新赋值prev和cursorthis.prev prev;this.cursor cursor;return success;
}
当我们想要从stack中获取对象时首先会从WeakOrderQueue链表中获取从上面代码也可以看到通过上面的代码可以看到cursor指针表示的是当前遍历到的WeakOrderQueue如果这个WeakOrderQueue中的对象已经获取完了那么就继续从下一个节点去获取并更新cursor指针那么这里又是怎样从WeakOrderQueue中获取对象的呢具体逻辑在transfer方法中
/*** 把当前WeakOrderQueue对象中的link链表的头节点中的全部可回收对象转移到指定的stack中* param dst 指定的stack* return true有转移回收对象 false没有转移回收对象*/
SuppressWarnings(rawtypes)
boolean transfer(Stack? dst) {Link head this.head.link;// 条件成立说明该WeakOrderQueue对象中还没有link链表if (head null) {return false;}// 条件成立说明当前头节点已经把回收对象全部转移完了if (head.readIndex LINK_CAPACITY) {// 如果头节点的下一个节点为null就说明没有回收对象可转移了if (head.next null) {return false;}// 把下一个节点设置为头节点head head.next;this.head.relink(head);}// 获取到当前的读指针final int srcStart head.readIndex;// 获取到这个link节点中存放的回收对象数量int srcEnd head.get();// 两者相减得到的就是需要转移的回收对象数量final int srcSize srcEnd - srcStart;// 条件成立说明没有可回收的对象能够转移if (srcSize 0) {return false;}// 获取到转移目标的stack中的存活对象数量final int dstSize dst.size;// 计算出当把可回收对象全部转移到stack中时stack所需的大小final int expectedCapacity dstSize srcSize;// 条件成立说明如果把可回收对象全部转移到stack中的时候stack需要扩容if (expectedCapacity dst.elements.length) {// 对stack进行扩容得到扩容后的stack大小final int actualCapacity dst.increaseCapacity(expectedCapacity);// actualCapacity - dstSize得到的就是stack扩容之后剩余的大小// 因为stack扩容之后有可能还是完全放不下link中可转移的回收对象所以这里取最小值srcEnd min(srcStart actualCapacity - dstSize, srcEnd);}// 条件成立说明link节点中存在可转移的回收对象if (srcStart ! srcEnd) {// 获取到link节点中的对象数组final DefaultHandle[] srcElems head.elements;// 获取到转移目标stack中的对象数组final DefaultHandle[] dstElems dst.elements;int newDstSize dstSize;// 遍历link节点的对象数组的可回收对象for (int i srcStart; i srcEnd; i) {DefaultHandle? element srcElems[i];if (element.recycleId 0) {element.recycleId element.lastRecycledId;} else if (element.recycleId ! element.lastRecycledId) {throw new IllegalStateException(recycled already);}srcElems[i] null;// 判断是否能够回收该对象if (dst.dropHandle(element)) {// 放弃回收该对象continue;}// 代码执行到这里说明该对象能够被回收// 重新给handle中的stack属性赋值element.stack dst;// 把对象放到stack的对象数组中此时就完成了该对象的回收dstElems[newDstSize ] element;}// 条件成立说明这个link已经把可回收对象全部转移完了此时把下一个link节点设置为头节点if (srcEnd LINK_CAPACITY head.next ! null) {// Add capacity back as the Link is GCed.this.head.relink(head.next);}// 把可回收的对象转移完了之后更新读指针head.readIndex srcEnd;// 条件成立说明此次没有转移到一个回收对象if (dst.size newDstSize) {return false;}// 代码执行到这里说明此次转移到至少一个回收对象此时更新stack的sizedst.size newDstSize;return true;} else {// The destination stack is full already.return false;}
}
我们从上面已经知道每一个WeakOrderQueue对象其实就是一个由link节点组成的链表而每一个link节点中存储对象靠的就是一个数组。当调用transfer方法的时候首先会获取当前link链表的头节点根据读指针和写指针计算出这个link节点转移对象的数组开始下标和结束下标然后根据开始下标和结束下标把数组中的对象转移到stack中最后判断这个link节点是否已经全部转移完里面的对象了如果已经转移完了就把下一个link节点置为头节点这样当下一次再调用transfer方法的时候就能够又能从头节点获取了。也就是说每调用一次transfer方法就会转移最多一个link节点数量的对象如果一个对象都没有转移到那么transfer方法就返回false反之返回true。
分析了transfer方法我们再回到scavengeSome方法中重新进行分析在scavengeSome方法中大概的执行流程如下
首先会通过cursor指针定位到已经遍历到的WeakOrderQueue对象然后调用这个WeakOrderQueue对象的transfer方法进行对象转移如果transfer方法返回true则跳出do...while循环反之如果返回false则获取下一个WeakOrderQueue并赋值给cursor指针然后do...while循环重复上面第一点的步骤如果transfer返回false则说明这一次没有转移到对象此时会再去判断当前这个WeakOrderQueue对应的线程是否已经挂了如果挂了则再判断WeakOrderQueue中是否还存在没有被转移的link节点因为有可能由于回收频率的控制导致这个link节点中本来有的对象被放弃回收了但是该节点后可能还存在其他没有被转移的link节点如果存在则把剩余的这些link节点中的对象全部转移到stack中最后再把这个WeakOrderQueue占用的回收对象数量归还给availableSharedCapacity属性并且把这个WeakOrderQueue从链表中删除
总结来说scavengeSome方法会去遍历WeakOrderQueue链表如果当前WeakOrderQueue转移不到对象就换下一个WeakOrderQueue节点进行转移当发现有WeakOrderQueue节点转移对象成功之后就返回true反之返回false
