通信双方建立TCP连接后，主动关闭连接的一方就会进入TIME_WAIT状态，TIME_WAIT状态维持时间是两个MSL时间长度，也就是在1-4分钟，Windows操作系统就是4分钟。进入TIME_WAIT状态的一般情况下是客户端，一个TIME_WAIT状态的连接就占用了一个本地端口。一台机器上端口号数量的上限是65536个，如果在同一台机器上进行压力测试模拟上万的客户请求，并且循环与服务端进行短连接通信，那么这台机器将产生4000个左右的TIME_WAIT Socket，后续的短连接就会产生address already in use : connect的异常，如果使用Nginx作为方向代理也需要考虑TIME_WAIT状态，发现系统存在大量TIME_WAIT状态的连接，通过调整内核参数解决。
/** * Generate an Object pool with a specified `factory` and `config`. * * @param {typeof DefaultEvictor} Evictor * @param {typeof Deque} Deque * @param {typeof PriorityQueue} PriorityQueue * @param {Object} factory * Factory to be used for generating and destroying the items. * @param {Function} factory.create * Should create the item to be acquired, * and call it's first callback argument with the generated item as it's argument. * @param {Function} factory.destroy * Should gently close any resources that the item is using. * Called before the items is destroyed. * @param {Function} factory.validate * Test if a resource is still valid .Should return a promise that resolves to a boolean, true if resource is still valid and false * If it should be removed from pool. * @param {Object} options */ constructor(Evictor, Deque, PriorityQueue, factory, options) { super(); factoryValidator(factory); // 检验我们定义的factory的有效性包含create destroy validate this._config = new PoolOptions(options); // 连接池配置 // TODO: fix up this ugly glue-ing this._Promise = this._config.Promise;
this._factory = factory; this._draining = false; this._started = false; /** * Holds waiting clients * @type {PriorityQueue} */ this._waitingClientsQueue = new PriorityQueue(this._config.priorityRange); // 请求的对象管管理队列queue 初始化queue的size 1 { _size: 1, _slots: [ Queue { _list: [Object] } ] } /** * Collection of promises for resource creation calls made by the pool to factory.create * @type {Set} */ this._factoryCreateOperations = new Set(); // 正在创建的长连接
/** * Collection of promises for resource destruction calls made by the pool to factory.destroy * @type {Set} */ this._factoryDestroyOperations = new Set(); // 正在销毁的长连接
/** * A queue/stack of pooledResources awaiting acquisition * TODO: replace with LinkedList backed array * @type {Deque} */ this._availableObjects = new Deque(); // 空闲的资源长连接
/** * Collection of references for any resource that are undergoing validation before being acquired * @type {Set} */ this._testOnBorrowResources = new Set(); // 正在检验有效性的资源
/** * Collection of references for any resource that are undergoing validation before being returned * @type {Set} */ this._testOnReturnResources = new Set();
/** * Collection of promises for any validations currently in process * @type {Set} */ this._validationOperations = new Set();// 正在校验的中间temp
/** * All objects associated with this pool in any state (except destroyed) * @type {Set} */ this._allObjects = new Set(); // 所有的链接资源 是一个 PooledResource对象
/** * Loans keyed by the borrowed resource * @type {Map} */ this._resourceLoans = new Map(); // 被借用的对象的map release的时候用到
/** * Infinitely looping iterator over available object * @type {DequeIterator} */ this._evictionIterator = this._availableObjects.iterator(); // 一个迭代器
this._evictor = new Evictor();
/** * handle for setTimeout for next eviction run * @type {(number|null)} */ this._scheduledEviction = null;
/** * Request a new resource. The callback will be called, * when a new resource is available, passing the resource to the callback. * TODO: should we add a seperate "acquireWithPriority" function * * @param {Number} [priority=0] * Optional. Integer between 0 and (priorityRange - 1). Specifies the priority * of the caller if there are no available resources. Lower numbers mean higher * priority. * * @returns {Promise} */ acquire(priority) { // 空闲资源队列资源是有优先等级的 if (this._started === false && this._config.autostart === false) { this.start(); // 会在this._allObjects 添加min的连接对象数 } if (this._draining) { // 如果是在资源释放阶段就不能再请求资源了 return this._Promise.reject( new Error("pool is draining and cannot accept work") ); } // 如果要设置了等待队列的长度且要等待 如果超过了就返回资源不可获取 // TODO: should we defer this check till after this event loop incase "the situation" changes in the meantime if ( this._config.maxWaitingClients !== undefined && this._waitingClientsQueue.length >= this._config.maxWaitingClients ) { return this._Promise.reject( new Error("max waitingClients count exceeded") ); }
return resourceRequest.promise; // 返回的是一个promise对象resolve却是在其他地方触发 } /** * Attempt to resolve an outstanding resource request using an available resource from * the pool, or creating new ones * * @private */ _dispense() { /** * Local variables for ease of reading/writing * these don't (shouldn't) change across the execution of this fn */ const numWaitingClients = this._waitingClientsQueue.length; // 正在等待的请求的队列长度 各个优先级的总和 console.log('numWaitingClients', numWaitingClients) // 1
// If there aren't any waiting requests then there is nothing to do // so lets short-circuit if (numWaitingClients < 1) { return; } // max: 10, min: 4 console.log('_potentiallyAllocableResourceCount', this._potentiallyAllocableResourceCount) // 目前潜在空闲可用的连接数量 const resourceShortfall = numWaitingClients - this._potentiallyAllocableResourceCount; // 还差几个可用的 小于零表示不需要 大于0表示需要新建长连接的数量 console.log('spareResourceCapacity', this.spareResourceCapacity) // 距离max数量的还有几个没有创建 const actualNumberOfResourcesToCreate = Math.min( this.spareResourceCapacity, // -6 resourceShortfall // 这个是 -3 ); // 如果resourceShortfall>0 表示需要新建但是这新建的数量不能超过spareResourceCapacity最多可创建的 console.log('actualNumberOfResourcesToCreate', actualNumberOfResourcesToCreate) // 如果actualNumberOfResourcesToCreate >0 表示需要创建连接 for (let i = 0; actualNumberOfResourcesToCreate > i; i++) { this._createResource(); // 新增新的长连接 }
// If we are doing test-on-borrow see how many more resources need to be moved into test // to help satisfy waitingClients if (this._config.testOnBorrow === true) { // 如果开启了使用前校验资源的有效性 // how many available resources do we need to shift into test const desiredNumberOfResourcesToMoveIntoTest = numWaitingClients - this._testOnBorrowResources.size;// 1 const actualNumberOfResourcesToMoveIntoTest = Math.min( this._availableObjects.length, // 3 desiredNumberOfResourcesToMoveIntoTest // 1 ); for (let i = 0; actualNumberOfResourcesToMoveIntoTest > i; i++) { // 需要有效性校验的数量 至少满足最小的waiting clinet this._testOnBorrow(); // 资源有效校验后再分发 } }
// if we aren't testing-on-borrow then lets try to allocate what we can if (this._config.testOnBorrow === false) { // 如果没有开启有效性校验 就开启有效资源的分发 const actualNumberOfResourcesToDispatch = Math.min( this._availableObjects.length, numWaitingClients ); for (let i = 0; actualNumberOfResourcesToDispatch > i; i++) { // 开始分发资源 this._dispatchResource(); } } } /** * Attempt to move an available resource to a waiting client * @return {Boolean} [description] */ _dispatchResource() { if (this._availableObjects.length < 1) { return false; }
const pooledResource = this._availableObjects.shift(); // 从可以资源池里面取出一个 this._dispatchPooledResourceToNextWaitingClient(pooledResource); // 分发 return false; } /** * Dispatches a pooledResource to the next waiting client (if any) else * puts the PooledResource back on the available list * @param {PooledResource} pooledResource [description] * @return {Boolean} [description] */ _dispatchPooledResourceToNextWaitingClient(pooledResource) { const clientResourceRequest = this._waitingClientsQueue.dequeue(); // 可能是undefined 取出一个等待的quene console.log('clientResourceRequest.state', clientResourceRequest.state); if (clientResourceRequest === undefined || clientResourceRequest.state !== Deferred.PENDING) { console.log('没有等待的') // While we were away either all the waiting clients timed out // or were somehow fulfilled. put our pooledResource back. this._addPooledResourceToAvailableObjects(pooledResource); // 在可用的资源里面添加一个 // TODO: do need to trigger anything before we leave? return false; } // TODO clientResourceRequest 的state是否需要判断 如果已经是resolve的状态 已经超时回去了 这个是否有问题 const loan = new ResourceLoan(pooledResource, this._Promise); this._resourceLoans.set(pooledResource.obj, loan); // _resourceLoans 是个map k=>value pooledResource.obj 就是socket本身 pooledResource.allocate(); // 标识资源的状态是正在被使用 clientResourceRequest.resolve(pooledResource.obj); // acquire方法返回的promise对象的resolve在这里执行的 return true; }
