将当前的符合条件的processIds和当前的reply queue进行校对，剔除不在processIds里的内容
961977
protected synchronized void compareReply(List<Long> processIds) {
        Object[] replyIds = replyProcessIds.toArray();
        for (Object replyId : replyIds) {
            if (processIds.contains((Long) replyId) == false) { // 判断reply id是否在当前processId列表中
                // 因为存在并发问题，如在执行Listener事件的同时，可能触发了process的创建，这时新建的processId会进入到reply队列中
                // 此时接受到的processIds变量为上一个版本的内容，所以会删除新建的process，导致整个通道被挂住
                if (CollectionUtils.isEmpty(processIds) == false) {
                    Long processId = processIds.get(0);
                    if (processId > (Long) replyId) { // 如果当前最小的processId都大于replyId, processId都是递增创建的
                        processTermined((Long) replyId); // 调用一下删除操作
                    }
                }
            }
        }
    }
1060887
def do_list_queue(self, line):
        

        def f(p, args):
            o = p.get()
            if o.resources.queue:
                for q in o.resources.queue:
                    print('%s %s' % (q.resource_id, q.port))

        self._request(line, f)
722200
def fetch_in_thread(sr, nsa):
    

    def fetch_seq(q, nsa):
        pid, ppid = os.getpid(), os.getppid()
        q.put((pid, ppid, sr[nsa]))
    
    q = Queue()
    p = Process(target=fetch_seq, args=(q, nsa))
    p.start()
    pid, ppid, seq = q.get()
    p.join()

    assert pid != ppid, "sequence was not fetched from thread"
    return pid, ppid, seq
1129805
private List<String> getProcessList(String queueName) throws RabbitmqCommunicateException
    {
        List<String> processList = getContextBuilder().getProcessList(queueName);
        if (processList == null || processList.size() == 0)
        {
            String messageFmt = "QueueName's ProcessList is not defined. QueueName={0}";
            String message = MessageFormat.format(messageFmt, queueName);
            throw new RabbitmqCommunicateException(message);
        }
        return processList;
    }
834706
function(threadId, active, callback) {
                if (active) {
                    if (hasCache && typeof queueDb == 'object') {
                        queueDb.waitQ.where('threadId')
                            .equals(threadId)
                            .and(function(item) {
                                return item.owner == parseInt(userInfo.id);
                            })
                            .toArray()
                            .then(function(waitQueueOnCache) {
                                var uniqueIds = [];

                                for (var i = 0; i < waitQueueOnCache.length; i++) {
                                    uniqueIds.push(waitQueueOnCache[i].uniqueId);
                                }

                                if (uniqueIds.length && chatState) {
                                    sendMessage({
                                        chatMessageVOType: chatMessageVOTypes.GET_HISTORY,
                                        content: {
                                            uniqueIds: uniqueIds
                                        },
                                        subjectId: threadId
                                    }, {
                                        onResult: function(result) {
                                            if (!result.hasError) {
                                                var messageContent = result.result;

                                                /**
                                                 * Delete those messages from wait
                                                 * queue which are already on the
                                                 * server databases
                                                 */
                                                for (var i = 0; i < messageContent.length; i++) {
                                                    for (var j = 0; j < uniqueIds.length; j++) {
                                                        if (uniqueIds[j] == messageContent[i].uniqueId) {
                                                            deleteFromChatWaitQueue(messageContent[i], function() {
                                                            });
                                                            uniqueIds.splice(j, 1);
                                                            waitQueueOnCache.splice(j, 1);
                                                        }
                                                    }
                                                }

                                                /**
                                                 * Delete those messages from wait
                                                 * queue which are in the send
                                                 * queue and are going to be sent
                                                 */
                                                for (var i = 0; i < chatSendQueue.length; i++) {
                                                    for (var j = 0; j < uniqueIds.length; j++) {
                                                        if (uniqueIds[j] == chatSendQueue[i].message.uniqueId) {
                                                            deleteFromChatWaitQueue(chatSendQueue[i].message, function() {
                                                            });
                                                            uniqueIds.splice(j, 1);
                                                            waitQueueOnCache.splice(j, 1);
                                                        }
                                                    }
                                                }

                                                callback && callback(waitQueueOnCache);
                                            }
                                        }
                                    });
                                }
                                else {
                                    callback && callback(waitQueueOnCache);
                                }
                            })
                            .catch(function(error) {
                                fireEvent('error', {
                                    code: error.code,
                                    message: error.message,
                                    error: error
                                });
                            });
                    }
                    else {
                        var uniqueIds = [];

                        for (var i = 0; i < chatWaitQueue.length; i++) {
                            uniqueIds.push(chatWaitQueue[i].uniqueId);
                        }

                        if (uniqueIds.length) {
                            sendMessage({
                                chatMessageVOType: chatMessageVOTypes.GET_HISTORY,
                                content: {
                                    uniqueIds: uniqueIds
                                },
                                subjectId: threadId
                            }, {
                                onResult: function(result) {
                                    if (!result.hasError) {
                                        var messageContent = result.result;

                                        for (var i = 0; i < messageContent.length; i++) {
                                            for (var j = 0; j < uniqueIds.length; j++) {
                                                if (uniqueIds[j] == messageContent[i].uniqueId) {
                                                    uniqueIds.splice(j, 1);
                                                    chatWaitQueue.splice(j, 1);
                                                }
                                            }
                                        }
                                        callback && callback(chatWaitQueue);
                                    }
                                }
                            });
                        }
                        else {
                            callback && callback([]);
                        }
                    }
                }
                else {
                    callback && callback([]);
                }
            }
206546
def get_processing_block_ids():
    """"""
    ids = []
    for key in sorted(DB.keys(pattern='scheduling_block/*')):
        config = json.loads(DB.get(key))
        for processing_block in config['processing_blocks']:
            ids.append(processing_block['id'])
    return ids
744534
private void recovery(Long pipelineId) {
        List<Long> currentProcessIds = stageMonitor.getCurrentProcessIds(false);
        for (Long processId : currentProcessIds) {
            String path = StagePathUtils.getProcess(pipelineId, processId);
            try {
                byte[] bytes = zookeeper.readData(path);
                ProcessNodeEventData nodeData = JsonUtils.unmarshalFromByte(bytes, ProcessNodeEventData.class);
                if (nodeData.getStatus().isUnUsed()) {// 加入未使用的processId
                    addReply(processId);
                }
            } catch (ZkException e) {
                logger.error("SelectStageListener", e);
            }
        }
    }
864214
async def append(self, reply: Reply) \
            -> None:
        
        result = reply.result
        identifier = result.get(f.IDENTIFIER.nm)
        txnId = result.get(TXN_ID)
        logger.debug("Reply being sent {}".format(reply))
        if self._isNewTxn(identifier, reply, txnId):
            self.addToProcessedTxns(identifier, txnId, reply)
        if identifier not in self.responses:
            self.responses[identifier] = asyncio.Queue()
        await self.responses[identifier].put(reply)
368788
def FetchAndRemoveRequestsAndResponses(self, session_id):
    """"""
    messages = []
    with queue_manager.WellKnownQueueManager(token=self.token) as manager:
      for response in manager.FetchResponses(session_id):
        messages.append(response)
      manager.DeleteWellKnownFlowResponses(session_id, messages)

    return messages
1177522
def do_processlist(self, arg):
        
        if self.cmdprefix:
            raise CmdError("prefix not allowed")
        if arg:
            raise CmdError("too many arguments")
        system   = self.debug.system
        pid_list = self.debug.get_debugee_pids()
        if pid_list:
            print("Process ID   File name")
            for pid in pid_list:
                if   pid == 0:
                    filename = "System Idle Process"
                elif pid == 4:
                    filename = "System"
                else:
                    filename = system.get_process(pid).get_filename()
                    filename = PathOperations.pathname_to_filename(filename)
                print("%-12d %s" % (pid, filename))
1367322
def get_processing_block_ids(self):
        """"""

        # Initialise empty list
        _processing_block_ids = []

        # Pattern used to search processing block ids
        pattern = '*:processing_block:*'
        block_ids = self._db.get_ids(pattern)

        for block_id in block_ids:
            id_split = block_id.split(':')[-1]
            _processing_block_ids.append(id_split)
        return sorted(_processing_block_ids)
169223
public boolean ack(Long processId) {
        boolean result = waitProcessIds.ack();
        if (logger.isDebugEnabled()) {
            logger.debug("## {} ack termin id [{}]", getPipelineId(), processId);
        }
        return result;
    }
1243486
public long getPID() {
		long currentPID = cachedPID.get();
		if (currentPID != 0) {
			return currentPID;
		}
		try {
			currentPID = detectPID();
		} catch (Throwable cause) {
			logger.info("Unable to detect process ID!", cause);
		}
		if (currentPID == 0) {
			currentPID = System.nanoTime();
			if (!cachedPID.compareAndSet(0, currentPID)) {
				currentPID = cachedPID.get();
			}
		} else {
			cachedPID.set(currentPID);
		}
		return currentPID;
	}
998616
function(id) {
        var q = queues[id], msg, sc;
        if (q.timer) {
            // q.timer.cancel();
            clearTimeout(q.timer);
        }
        q.finished = true;

        if (q.aborted) {
            msg = "transaction " + id + " was aborted";
            _fail(id, msg);
            return;
        }

        // execute success callback
        if (q.onSuccess) {
            sc = q.context || q;
            q.onSuccess.call(sc, _returnData(q));
        }

        _end(id, msg, 'OK');
    }
612176
public static String getProcess(Long pipelineId, Long processId) {
        // 根据channelId , pipelineId构造path
        return MessageFormat.format(ArbitrateConstants.NODE_PROCESS_FORMAT, getChannelId(pipelineId),
                                    String.valueOf(pipelineId), getProcessNode(processId));
    }
66394
public static String getProcessNode(Long processId) {
        return StringUtils.leftPad(String.valueOf(processId.intValue()), 10, '0');
    }
653427
protected void process(DBIDRef id, ArrayDBIDs ids, DBIDArrayIter it, int n, WritableDBIDDataStore pi, WritableDoubleDataStore lambda, WritableDoubleDataStore m) {
    slinkstep3(id, it, n, pi, lambda, m);
    slinkstep4(id, it, n, pi, lambda);
  }