汇总各个Worker的运行状态信息
773916
public function stats()
    {
        $data = [
            'worker' => [
                // worker进程ID
                // 'pid' => 0,
                // 协程统计信息
                // 'coroutine' => [
                // 当前正在处理的请求数
                // 'total' => 0,
                //],
                // 内存使用
                // 'memory' => [
                // 峰值
                // 'peak'  => '',
                // 当前使用
                // 'usage' => '',
                //],
                // 请求信息
                //'request' => [
                // 当前Worker进程收到的请求次数
                //'worker_request_count' => 0,
                //],
            ],
            'tcp' => [
                // 服务器启动的时间
                'start_time' => '',
                // 当前连接的数量
                'connection_num' => 0,
                // 接受了多少个连接
                'accept_count' => 0,
                // 关闭的连接数量
                'close_count' => 0,
                // 当前正在排队的任务数
                'tasking_num' => 0,
                // Server收到的请求次数
                'request_count' => 0,
                // 消息队列中的Task数量
                'task_queue_num' => 0,
                // 消息队列的内存占用字节数
                'task_queue_bytes' => 0,
            ],
        ];

        $workerIds = range(0, $this->MSFServer->server->setting['worker_num'] - 1);
        foreach ($workerIds as $workerId) {
            $workerInfo = @$this->MSFServer->sysCache->get(Macro::SERVER_STATS . $workerId);
            if ($workerInfo) {
                $data['worker']['worker' . $workerId] = $workerInfo;
            } else {
                $data['worker']['worker' . $workerId] = [];
            }
        }

        $lastStats = $this->MSFServer->sysCache->get(Macro::SERVER_STATS);
        $data['tcp'] = $this->MSFServer->server->stats();
        $data['running']['qps'] = $data['tcp']['request_count'] - $lastStats['tcp']['request_count'];

        if (!isset($lastStats['running']['last_qpm'])) {
            $data['running']['last_qpm'] = 0;
        } else {
            $data['running']['last_qpm'] = $lastStats['running']['last_qpm'];
        }

        if ($this->lastMinute >= ceil(time() / 60)) {
            if (!empty($lastStats) && !isset($lastStats['running']['qpm'])) {
                $lastStats['running']['qpm'] = 0;
            }

            $data['running']['qpm'] = $lastStats['running']['qpm'] + $data['running']['qps'];
        } else {
            if (!empty($lastStats['running']['qpm'])) {
                $data['running']['last_qpm'] = $lastStats['running']['qpm'];
            }
            $data['running']['qpm'] = $data['running']['qps'];
            $this->lastMinute = ceil(time() / 60);
        }

        unset($data['tcp']['worker_request_count']);
        $this->MSFServer->sysCache->set(Macro::SERVER_STATS, $data);
    }
316372
function finish(\Swoole\Server $server, int $taskId, string $data){//仅限worker进程中触发
    echo __METHOD__, $taskId, $data, PHP_EOL;
  }
500500
public function sendToOneWorker($workerId, $type, $uns_data, string $callStaticFuc)
    {
        $send_data = get_instance()->packServerMessageBody($type, $uns_data, $callStaticFuc);
        if ($this->server->worker_id == $workerId) {
            //自己的进程是收不到消息的所以这里执行下
            sd_call_user_func($callStaticFuc, $uns_data);
        } else {
            get_instance()->server->sendMessage($send_data, $workerId);
        }
    }
254353
def worker_status(obj, filter_queues, details):
    
    show_colors = obj['show_color']

    f_queues = filter_queues.split(',') if filter_queues is not None else None

    workers = list_workers(config=obj['config'], filter_by_queues=f_queues)
    if len(workers) == 0:
        click.echo('No workers are running at the moment.')
        return

    for ws in workers:
        click.echo('{} {}'.format(_style(show_colors, 'Worker:', fg='blue', bold=True),
                                  _style(show_colors, ws.name, fg='blue')))
        click.echo('{:23} {}'.format(_style(show_colors, '> pid:', bold=True), ws.pid))

        if details:
            click.echo('{:23} {}'.format(_style(show_colors, '> concurrency:', bold=True),
                                         ws.concurrency))
            click.echo('{:23} {}'.format(_style(show_colors, '> processes:', bold=True),
                                         ', '.join(str(p) for p in ws.process_pids)))
            click.echo('{:23} {}://{}:{}/{}'.format(_style(show_colors, '> broker:',
                                                           bold=True),
                                                    ws.broker.transport,
                                                    ws.broker.hostname,
                                                    ws.broker.port,
                                                    ws.broker.virtual_host))

        click.echo('{:23} {}'.format(_style(show_colors, '> queues:', bold=True),
                                     ', '.join([q.name for q in ws.queues])))

        if details:
            click.echo('{:23} {}'.format(_style(show_colors, '> job count:', bold=True),
                                         ws.job_count))

            jobs = list_jobs(config=obj['config'], filter_by_worker=ws.name)
            click.echo('{:23} [{}]'.format(_style(show_colors, '> jobs:', bold=True),
                                           len(jobs) if len(jobs) > 0 else 'No tasks'))

            for job in jobs:
                click.echo('{:15} {} {}'.format(
                    '',
                    _style(show_colors, '{}'.format(job.name),
                           bold=True, fg=JOB_COLOR[job.type]),
                    _style(show_colors, '({}) [{}] <{}> on {}'.format(
                        job.type, job.workflow_id, job.id, job.worker_pid),
                        fg=JOB_COLOR[job.type])))

        click.echo('\n')
64138
function createWorkerExitHandler(port) {
        return function (code, signal) {
            var workerIndex = port - startPort;

            workers[workerIndex] = null;

            if (code !== 0 || code === null) {
                console.log('Worker exited with code: ' + code);

                if (!shuttingDown) {
                    // Start worker again after 1 sec
                    setTimeout(function () {
                        if (!workers[workerIndex]) {
                            workers[workerIndex] = startWorker(port);
                        }
                    }, 1000);
                }
            }
        };
    }
367467
protected function onWorkerReload($worker)
    {
        // 防止进程立刻退出
        $worker->reloadable = false;
        // 延迟 0.05 秒退出，避免 BusinessWorker 瞬间全部退出导致没有可用的 BusinessWorker 进程
        Timer::add(0.05, array('Workerman\Worker', 'stopAll'));
        // 执行用户定义的 onWorkerReload 回调
        if ($this->_onWorkerReload) {
            call_user_func($this->_onWorkerReload, $this);
        }
    }
583510
def worker(self, fifo):
        '''
        '''
        logging.debug('Starting worker thread for %s', fifo)
        self._fifo_conditions[fifo].acquire()
        while True:
            try:
                data_tuple = self._fifo_data_deque[fifo].popleft()
            except IndexError:
                self._fifo_conditions[fifo].wait(self.readout_interval)  # sleep a little bit, reducing CPU usage
            else:
                if data_tuple is None:  # if None then exit
                    break
                else:
                    for index, (filter_func, converter_func, fifo_select) in enumerate(izip(self.filter_func, self.converter_func, self.fifo_select)):
                        if fifo_select is None or fifo_select == fifo:
                            # filter and do the conversion
                            converted_data_tuple = convert_data_iterable((data_tuple,), filter_func=filter_func, converter_func=converter_func)[0]
                            n_data_words = converted_data_tuple[0].shape[0]
                            with self.data_words_per_second_lock:
                                self._words_per_read[index].append((n_data_words, converted_data_tuple[1], converted_data_tuple[2]))
                            self._data_deque[index].append(converted_data_tuple)
                            with self._data_conditions[index]:
                                self._data_conditions[index].notify_all()
        for index, fifo_select in enumerate(self.fifo_select):
            if fifo_select is None or fifo_select == fifo:
                self._data_deque[index].append(None)
                with self._data_conditions[index]:
                    self._data_conditions[index].notify_all()
        self._fifo_conditions[fifo].release()
        logging.debug('Stopping worker thread for %s', fifo)
1155756
public function isUserProcess($worker_id = null)
    {
        //一定要===
        if ($worker_id === null) {
            $worker_id = $this->workerId;
        }
        return $worker_id >= $this->worker_num + $this->task_num ? true : false;
    }
14469
function UserInfo() {
	//使用者帳號(索引用)
	this.strAccount = "";
	//電子信箱
	this.strEmail = "";
	//信箱通知勾選
	this.bEmailNotify = false;
	//電話資料
	this.strPhone = "";
	//簡訊通知勾選
	this.bSMSNotify = false;
	//使用者名稱(暱稱?)
	this.strUserName = "";
	//語系控制
	this.strLanguages = "";
	//系統保護
	this.bSystemAegis = false;
	//登入次數
	this.iLoginCount = 0;
	//登入時間
	this.iLogin = 0;
	//最後登出時間
	this.iLastLogout = 0;
}
278953
protected function onWorkerStart()
    {
        if (!class_exists('\Protocols\GatewayProtocol')) {
            class_alias('GatewayWorker\Protocols\GatewayProtocol', 'Protocols\GatewayProtocol');
        }

        if (!is_array($this->registerAddress)) {
            $this->registerAddress = array($this->registerAddress);
        }
        $this->connectToRegister();

        \GatewayWorker\Lib\Gateway::setBusinessWorker($this);
        \GatewayWorker\Lib\Gateway::$secretKey = $this->secretKey;
        if ($this->_onWorkerStart) {
            call_user_func($this->_onWorkerStart, $this);
        }
        
        if (is_callable($this->eventHandler . '::onWorkerStart')) {
            call_user_func($this->eventHandler . '::onWorkerStart', $this);
        }

        if (function_exists('pcntl_signal')) {
            // 业务超时信号处理
            pcntl_signal(SIGALRM, array($this, 'timeoutHandler'), false);
        } else {
            $this->processTimeout = 0;
        }

        // 设置回调
        if (is_callable($this->eventHandler . '::onConnect')) {
            $this->_eventOnConnect = $this->eventHandler . '::onConnect';
        }

        if (is_callable($this->eventHandler . '::onMessage')) {
            $this->_eventOnMessage = $this->eventHandler . '::onMessage';
        } else {
            echo "Waring: {$this->eventHandler}::onMessage is not callable\n";
        }

        if (is_callable($this->eventHandler . '::onClose')) {
            $this->_eventOnClose = $this->eventHandler . '::onClose';
        }

        if (is_callable($this->eventHandler . '::onWebSocketConnect')) {
            $this->_eventOnWebSocketConnect = $this->eventHandler . '::onWebSocketConnect';
        }

    }
1130493
function getWorkerData(worker) {
  /* istanbul ignore if */
  if (!worker) {
    throw new TypeError("Trying to get private data for null worker?!");
  }

  var data = worker[clusterphone.workerDataAccessor];
  if (!data) {
    worker[clusterphone.workerDataAccessor] = data = {};
  }

  return data;
}
1351167
def _get_worker_id(self, conn):
        """"""
        if self._worker_id is None:
            self._worker_id = conn.incr(self._key_worker())
        return self._worker_id
1057276
public function onWorkerError($serv, $worker_id, $worker_pid, $exit_code) {
        $this->access_log('worker_id = ' . $worker_id . '异常错误，pid=' . $worker_pid . '; exit_code=' . $exit_code);
    }
453105
def _worker_ctl(worker, lbn, vwa, profile='default'):
    '''
    
    '''

    cmd = {
        'cmd': 'update',
        'mime': 'prop',
        'w': lbn,
        'sw': worker,
        'vwa': vwa,
    }
    return _do_http(cmd, profile)['worker.result.type'] == 'OK'
796520
def go(self):
        
        myandro = self.settings["my"]

        def worker(idx, q):
            """
            Worker Thread
            """
            l.debug("Running worker-%d" % idx)

            while True:
                a, d, dx, axmlobj, arscobj = None, None, None, None, None
                try:
                    filename, fileraw = q.get()
                    id_file = zlib.adler32(fileraw)

                    l.debug("(worker-%d) get %s %d" % (idx, filename, id_file))

                    # FIXME: If the try/catch crashes before this line, there
                    # will be no logf to put into finish.
                    logf = self.settings["log"](id_file, filename)

                    is_analysis_dex, is_analysis_adex = True, True
                    l.debug("(worker-%d) filtering file %d" % (idx, id_file))
                    # TODO: This information should probably also go into the logf?
                    filter_file_ret, filter_file_type = myandro.filter_file(logf, fileraw)

                    if filter_file_ret:
                        l.debug("(worker-%d) analysis %s" % (id_file, filter_file_type))

                        if filter_file_type == "APK":
                            a = myandro.create_apk(logf, fileraw)
                            is_analysis_dex = myandro.analysis_apk(logf, a)
                            # TODO: Support multidex here
                            fileraw = a.get_dex()
                            filter_file_type = androconf.is_android_raw(fileraw)

                        elif filter_file_type == "AXML":
                            axmlobj = myandro.create_axml(logf, fileraw)
                            # TODO: the return value of analysis_axml is not checked
                            myandro.analysis_axml(logf, axmlobj)

                        elif filter_file_type == "ARSC":
                            arscobj = myandro.create_arsc(logf, fileraw)
                            # TODO: the return value of analysis_arsc is not checked
                            myandro.analysis_arsc(logf, arscobj)

                        if is_analysis_dex and filter_file_type == "DEX":
                            d = myandro.create_dex(logf, fileraw)
                            is_analysis_adex = myandro.analysis_dex(logf, d)

                        elif is_analysis_dex and filter_file_type == "DEY":
                            d = myandro.create_dey(logf, fileraw)
                            is_analysis_adex = myandro.analysis_dey(logf, d)

                        if is_analysis_adex and d:
                            # TODO: Support multidex here
                            dx = myandro.create_adex(logf, d)
                            # TODO: The return value of analysis_adex is not checked
                            myandro.analysis_adex(logf, dx)

                        # TODO: this is called also if AXML or ARSC is set...
                        myandro.analysis_app(logf, a, d, dx)
                        myandro.finish(logf)

                except Exception as why:
                    myandro.crash(logf, why)
                    # FIXME: finish is called here in any case of an exception
                    # but is only called if filter_file_ret is true above.
                    myandro.finish(logf)

                del a, d, dx, axmlobj, arscobj
                q.task_done()

        q = queue.Queue(self.settings["max_fetcher"])

        for i in range(self.settings["max_fetcher"]):
            t = threading.Thread(target=worker, args=[i, q])
            t.daemon = True
            t.start()

        # FIXME: Busy waiting with sleep...
        terminated = True
        while terminated:
            terminated = myandro.fetcher(q)

            try:
                if terminated:
                    time.sleep(10)
            except KeyboardInterrupt:
                terminated = False

        q.join()
67426
def _worker_method(self):
        
        while not self._should_quit():
            try:
                task = self._inq.get(block=True, timeout=0.25)
            except TypeError:
                if self._should_quit():
                    break
                else:
                    raise
            except Empty:
                continue

            try:
                obj = task.object
                if isinstance(obj, RiakObject):
                    rv = task.client.put(obj, **task.options)
                elif isinstance(obj, TsObject):
                    rv = task.client.ts_put(obj, **task.options)
                else:
                    raise ValueError('unknown obj type: %s'.format(type(obj)))
                task.outq.put(rv)
            except KeyboardInterrupt:
                raise
            except Exception as err:
                errdata = (task.object, err)
                task.outq.put(errdata)
            finally:
                self._inq.task_done()
184810
def workers(self):
        """"""
        worker_keys = self.redis_client.keys("Worker*")
        workers_data = {}

        for worker_key in worker_keys:
            worker_info = self.redis_client.hgetall(worker_key)
            worker_id = binary_to_hex(worker_key[len("Workers:"):])

            workers_data[worker_id] = {
                "node_ip_address": decode(worker_info[b"node_ip_address"]),
                "plasma_store_socket": decode(
                    worker_info[b"plasma_store_socket"])
            }
            if b"stderr_file" in worker_info:
                workers_data[worker_id]["stderr_file"] = decode(
                    worker_info[b"stderr_file"])
            if b"stdout_file" in worker_info:
                workers_data[worker_id]["stdout_file"] = decode(
                    worker_info[b"stdout_file"])
        return workers_data