发送给指定进程.

@param $workerId
@param $type
@param $uns_data
@param string $callStaticFuc
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);
        }
    }
639636
public function callActorBack($workerId, $token, $result, $node_name)
    {
        if ($node_name) {
            if ($node_name == $this->node_name) {//本机
                EventDispatcher::getInstance()->dispathToWorkerId($workerId, $token, $result);
                return;
            }
            if (!isset($this->client[$node_name])) return;
            $this->client[$node_name]->callActorBack($workerId, $token, $result);
        }
    }
120906
protected function changeWorkerState(string $workerId, int $workerState) : bool
    {
        $this->workerStates[$workerId] = $workerState;
        if ($workerState === Worker::WORKER_STATE_IDLE) {
            $this->pushJobs();
        }
        return true;
    }
260854
public function onWorkerError(\Swoole\Server $server, int $worker_id, int $worker_pid, int $exit_code)
    {
        echo date('Y-m-d H:i:s') . " |\033[31m http worker process error[id={$worker_id} pid={$worker_pid}] ......\033[0m \n";

        // Add event Listener
        $this->add_event_listener('onWorkerError');
    }
441430
public function initAsynPools($workerId)
    {
        $this->asynPools = [];
        if ($this->config->get('redis.enable', true)) {
            $this->addAsynPool('redisPool', new RedisAsynPool($this->config, $this->config->get('redis.active')));
        }
        if ($this->config->get('mysql.enable', true)) {
            $this->addAsynPool('mysqlPool', new MysqlAsynPool($this->config, $this->config->get('mysql.active')));
        }
        if ($this->config->get('error.dingding_enable', false)) {
            $this->addAsynPool('dingdingRest', new HttpClientPool($this->config, $this->config->get('error.dingding_url')));
        }
        $this->redis_pool = $this->asynPools['redisPool'] ?? null;
        $this->mysql_pool = $this->asynPools['mysqlPool'] ?? null;
    }
134575
public function onWorkerStart($serv, $workerId)
    {
        $this->initAsynPools();
        $this->initRedisProxies();
        //注册
        $this->asynPoolManager = new AsynPoolManager(null, $this);
        $this->asynPoolManager->noEventAdd();
        foreach ($this->asynPools as $pool) {
            if ($pool) {
                $pool->workerInit($workerId);
                $this->asynPoolManager->registerAsyn($pool);
            }
        }

        //redis proxy监测
        getInstance()->sysTimers[] = swoole_timer_tick(5000, function () {
            if (!empty($this->redisProxyManager)) {
                foreach ($this->redisProxyManager as $proxy) {
                    $proxy->check();
                }
            }
        });
    }
945851
private function handleProcessExitStatus($PID, $child_type, $exitStatus) {
        $child_type = $child_type === self::PROCESS_TYPE_JOB ? "child" : "watchdog";

        if ($exitStatus === 0) {
            $this->logger->debug("{type}: {$child_type} process exited successfully", $this->logContext);

            return false;
        }

        $jobFailedMessage = $this->readErrorFromSocket($this->sockets[$PID]) ?: "{$child_type} process failed with status: {$exitStatus}";
        $this->logger->error("{type}: fatal error in {$child_type} process: {$jobFailedMessage}", $this->logContext);

        return $jobFailedMessage;
    }
1190294
public function registerUrlFromStatus(\Vatsimphp\Sync\StatusSync $sync, $type)
    {
        $urls = $sync->loadData()->get($type)->toArray();
        if (empty($urls)) {
            throw new RuntimeException(
                'Error loading urls from StatusSync'
            );
        }
        $this->registerUrl($urls, true);

        return true;
    }
821808
private function setWorkerID($workerID)
    {
        if ($workerID >= 1 &&
            $workerID <= 255
        ) {
            return (int)$workerID;
        } elseif ($workerID !== null) {
            throw new \InvalidArgumentException('WorkerID out of range (1-255)');
        }
        return rand(1, 255);
    }
643955
async def workerTypeState(self, *args, **kwargs):
        

        return await self._makeApiCall(self.funcinfo["workerTypeState"], *args, **kwargs)
555846
function reserveWorker(workerType, callback) {
        var debug = false,
            execArgv = process.execArgv.filter(function (arg) {
                if (arg.indexOf('--debug-brk') === 0) {
                    debug = '--debug-brk';
                } else if (arg.indexOf('--debug') === 0) {
                    debug = '--debug';
                } else if (arg.indexOf('--inspect-brk') === 0) {
                    debug = '--inspect-brk';
                } else if (arg.indexOf('--inspect') === 0) {
                    debug = '--inspect';
                } else {
                    return true;
                }

                return false;
            }),
            childProcess;

        // http://stackoverflow.com/questions/16840623/how-to-debug-node-js-child-forked-process
        // For some reason --debug-brk is given here..
        if (debug) {
            execArgv.push(debug + '=' + debugPort.toString());
            logger.info('Child debug port: ' + debugPort.toString());
            debugPort += 2;
        }

        logger.debug('execArgv for main process', process.execArgv);
        logger.debug('execArgv for new child process', execArgv);

        childProcess = Child.fork(CONNECTED_WORKER_JS, [], {execArgv: execArgv});

        _workers[childProcess.pid] = {
            childProcess: childProcess,
            state: CONSTANTS.workerStates.initializing,
            type: workerType,
            cb: null
        };

        logger.debug('workerPid forked ' + childProcess.pid);
        childProcess.on('message', function (msg) {
            messageHandling(msg);

            // FIXME: Do we really need the "initialize" in addition to "initialized"?
            // FIXME: Why couldn't the worker start the initializing at spawn? (It can load the gmeConfig itself)
            if (msg.type === CONSTANTS.msgTypes.initialized && typeof callback === 'function') {
                callback();
                callback = null;
            }
        });

        childProcess.on('exit', function (code, signal) {
            logger.debug('worker has exited: ' + childProcess.pid);
            // When killing child-process the code is undefined and the signal SIGINT.
            if (code !== 0 && signal !== 'SIGINT') {
                logger.warn('worker ' + childProcess.pid + ' has exited abnormally with code ' + code +
                    ', signal', signal);
                if (typeof callback === 'function') {
                    callback(new Error('worker ' + childProcess.pid + ' exited abnormally with code ' + code));
                }
            } else {
                logger.debug('worker ' + childProcess.pid + ' was terminated.');
            }

            if (workerType === CONSTANTS.workerTypes.connected) {
                self.connectedWorkerId = null;
            }

            delete _workers[childProcess.pid];
            reserveWorker(workerType);
        });
    }
843881
function WorkerPouch(opts, callback) {
  var api = this;

  if (typeof opts === 'string') {
    var slashIdx = utils.lastIndexOf(opts, '/');
    opts = {
      url: opts.substring(0, slashIdx),
      name: opts.substring(slashIdx + 1)
    };
  } else {
    opts = utils.clone(opts);
  }

  log('constructor called', opts);

  // Aspirational. once https://github.com/pouchdb/pouchdb/issues/5200
  // is resolved, you'll be able to directly pass in a worker here instead of
  // a function that returns a worker.
  var worker = (opts.worker && typeof opts.worker === 'function') ?
    opts.worker() : opts.worker;
  if (!worker || (!worker.postMessage && (!worker.controller || !worker.controller.postMessage))) {
    var workerOptsErrMessage =
      'Error: you must provide a valid `worker` in `new PouchDB()`';
    console.error(workerOptsErrMessage);
    return callback(new Error(workerOptsErrMessage));
  }

  if (!opts.name) {
    var optsErrMessage = 'Error: you must provide a database name.';
    console.error(optsErrMessage);
    return callback(new Error(optsErrMessage));
  }

  function handleUncaughtError(content) {
    try {
      api.emit('error', content);
    } catch (err) {
      // TODO: it's weird that adapters should have to handle this themselves
      console.error(
        'The user\'s map/reduce function threw an uncaught error.\n' +
        'You can debug this error by doing:\n' +
        'myDatabase.on(\'error\', function (err) { debugger; });\n' +
        'Please double-check your map/reduce function.');
      console.error(content);
    }
  }

  function onReceiveMessage(message) {
    var messageId = message.messageId;
    var messageType = message.type;
    var content = message.content;

    if (messageType === 'uncaughtError') {
      handleUncaughtError(content);
      return;
    }

    var cb = api._callbacks[messageId];

    if (!cb) {
      log('duplicate message (ignoring)', messageId, messageType, content);
      return;
    }

    log('receive message', api._instanceId, messageId, messageType, content);

    if (messageType === 'error') {
      delete api._callbacks[messageId];
      cb(content);
    } else if (messageType === 'success') {
      delete api._callbacks[messageId];
      cb(null, content);
    } else { // 'update'
      api._changesListeners[messageId](content);
    }
  }

  function workerListener(e) {
    if (e.data.id === api._instanceId) {
      onReceiveMessage(e.data);
    }
  }

  function postMessage(message) {
    /* istanbul ignore if */
    if (typeof worker.controller !== 'undefined') {
      // service worker, use MessageChannels because e.source is broken in Chrome < 51:
      // https://bugs.chromium.org/p/chromium/issues/detail?id=543198
      var channel = new MessageChannel();
      channel.port1.onmessage = workerListener;
      worker.controller.postMessage(message, [channel.port2]);
    } else {
      // web worker
      worker.postMessage(message);
    }
  }

  function sendMessage(type, args, callback) {
    if (api._destroyed) {
      return callback(new Error('this db was destroyed'));
    } else if (api._closed) {
      return callback(new Error('this db was closed'));
    }
    var messageId = uuid();
    log('send message', api._instanceId, messageId, type, args);
    api._callbacks[messageId] = callback;
    var encodedArgs = encodeArgs(args);
    postMessage({
      id: api._instanceId,
      type: type,
      messageId: messageId,
      args: encodedArgs
    });
    log('message sent', api._instanceId, messageId);
  }

  function sendRawMessage(messageId, type, args) {
    log('send message', api._instanceId, messageId, type, args);
    var encodedArgs = encodeArgs(args);
    postMessage({
      id: api._instanceId,
      type: type,
      messageId: messageId,
      args: encodedArgs
    });
    log('message sent', api._instanceId, messageId);
  }

  api.type = function () {
    return 'worker';
  };

  api._remote = false;

  api._id = adapterFun('id', function (callback) {
    sendMessage('id', [], callback);
  });

  api.compact = adapterFun('compact', function (opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('compact', [opts], callback);
  });

  api._info = function (callback) {
    sendMessage('info', [], callback);
  };

  api.get = adapterFun('get', function (id, opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('get', [id, opts], callback);
  });

  // hacky code necessary due to implicit breaking change in
  // https://github.com/pouchdb/pouchdb/commits/0ddeae6b
  api._get = function (id, opts, callback) {
    api.get(id, opts, function (err, doc) {
      if (err) {
        return callback(err);
      }
      callback(null, {doc: doc});
    });
  };

  api.remove =
    adapterFun('remove', function (docOrId, optsOrRev, opts, callback) {
      var doc;
      if (typeof optsOrRev === 'string') {
        // id, rev, opts, callback style
        doc = {
          _id: docOrId,
          _rev: optsOrRev
        };
        if (typeof opts === 'function') {
          callback = opts;
          opts = {};
        }
      } else {
        // doc, opts, callback style
        doc = docOrId;
        if (typeof optsOrRev === 'function') {
          callback = optsOrRev;
          opts = {};
        } else {
          callback = opts;
          opts = optsOrRev;
        }
      }
      var rev = (doc._rev || opts.rev);

      sendMessage('remove', [doc._id, rev], callback);
  });

  api.getAttachment =
    adapterFun('getAttachment', function (docId, attachmentId, opts,
                                                callback) {
      if (typeof opts === 'function') {
        callback = opts;
        opts = {};
      }
      sendMessage('getAttachment', [docId, attachmentId, opts], callback);
  });

  api.removeAttachment =
    adapterFun('removeAttachment', function (docId, attachmentId, rev,
                                                   callback) {

      sendMessage('removeAttachment', [docId, attachmentId, rev], callback);
    });

  // Add the attachment given by blob and its contentType property
  // to the document with the given id, the revision given by rev, and
  // add it to the database given by host.
  api.putAttachment =
    adapterFun('putAttachment', function (docId, attachmentId, rev, blob,
                                                type, callback) {
      if (typeof type === 'function') {
        callback = type;
        type = blob;
        blob = rev;
        rev = null;
      }
      if (typeof type === 'undefined') {
        type = blob;
        blob = rev;
        rev = null;
      }

      if (typeof blob === 'string') {
        var binary;
        try {
          binary = atob(blob);
        } catch (err) {
          // it's not base64-encoded, so throw error
          return callback(errors.error(errors.BAD_ARG,
            'Attachments need to be base64 encoded'));
        }
        blob = utils.createBlob([utils.binaryStringToArrayBuffer(binary)], {type: type});
      }

      var args = [docId, attachmentId, rev, blob, type];
      sendMessage('putAttachment', args, callback);
    });

  api.put = adapterFun('put', utils.getArguments(function (args) {
    var temp, temptype, opts;
    var doc = args.shift();
    var id = '_id' in doc;
    var callback = args.pop();
    if (typeof doc !== 'object' || Array.isArray(doc)) {
      return callback(errors.error(errors.NOT_AN_OBJECT));
    }

    doc = utils.clone(doc);

    preprocessAttachments(doc).then(function () {
      while (true) {
        temp = args.shift();
        temptype = typeof temp;
        if (temptype === "string" && !id) {
          doc._id = temp;
          id = true;
        } else if (temptype === "string" && id && !('_rev' in doc)) {
          doc._rev = temp;
        } else if (temptype === "object") {
          opts = utils.clone(temp);
        }
        if (!args.length) {
          break;
        }
      }
      opts = opts || {};

      sendMessage('put', [doc, opts], callback);
    }).catch(callback);

  }));

  api.post = adapterFun('post', function (doc, opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    opts = utils.clone(opts);

    sendMessage('post', [doc, opts], callback);
  });

  api._bulkDocs = function (req, opts, callback) {
    sendMessage('bulkDocs', [req, opts], callback);
  };

  api._allDocs = function (opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('allDocs', [opts], callback);
  };

  api._changes = function (opts) {
    opts = utils.clone(opts);

    if (opts.continuous) {
      var messageId = uuid();
      api._changesListeners[messageId] = opts.onChange;
      api._callbacks[messageId] = opts.complete;
      sendRawMessage(messageId, 'liveChanges', [opts]);
      return {
        cancel: function () {
          sendRawMessage(messageId, 'cancelChanges', []);
        }
      };
    }

    sendMessage('changes', [opts], function (err, res) {
      if (err) {
        opts.complete(err);
        return callback(err);
      }
      res.results.forEach(function (change) {
        opts.onChange(change);
      });
      if (opts.returnDocs === false || opts.return_docs === false) {
        res.results = [];
      }
      opts.complete(null, res);
    });
  };

  // Given a set of document/revision IDs (given by req), tets the subset of
  // those that do NOT correspond to revisions stored in the database.
  // See http://wiki.apache.org/couchdb/HttpPostRevsDiff
  api.revsDiff = adapterFun('revsDiff', function (req, opts, callback) {
    // If no options were given, set the callback to be the second parameter
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }

    sendMessage('revsDiff', [req, opts], callback);
  });

  api._query = adapterFun('query', function (fun, opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    var funEncoded = fun;
    if (typeof fun === 'function') {
      funEncoded = {map: fun};
    }
    sendMessage('query', [funEncoded, opts], callback);
  });

  api._viewCleanup = adapterFun('viewCleanup', function (callback) {
    sendMessage('viewCleanup', [], callback);
  });

  api._close = function (callback) {
    api._closed = true;
    callback();
  };

  api.destroy = adapterFun('destroy', function (opts, callback) {
    if (typeof opts === 'function') {
      callback = opts;
      opts = {};
    }
    sendMessage('destroy', [], function (err, res) {
      if (err) {
        api.emit('error', err);
        return callback(err);
      }
      api._destroyed = true;
      worker.removeEventListener('message', workerListener);
      api.emit('destroyed');
      callback(null, res);
    });
  });

  // api.name was added in pouchdb 6.0.0
  api._instanceId = api.name || opts.originalName;
  api._callbacks = {};
  api._changesListeners = {};

  worker.addEventListener('message', workerListener);

  var workerOpts = {
    name: api._instanceId,
    auto_compaction: !!opts.auto_compaction,
    storage: opts.storage
  };
  if (opts.revs_limit) {
    workerOpts.revs_limit = opts.revs_limit;
  }

  sendMessage('createDatabase', [workerOpts], function (err) {
    if (err) {
      return callback(err);
    }
    callback(null, api);
  });
}
821803
function Worker(baseUrl, opts = {}) {

  if (!(this instanceof Worker)) {
    return new Worker(baseUrl, opts);
  }

  // inheritance
  Emitter.call(this);

  this.options = extend({
    workerId: uuid.v4()
  }, defaultOptions, opts);

  this.subscriptions = {};

  this.state = STATE_NEW;

  // apply extensions
  if (this.options.use) {
    this.extend(this.options.use);
  }

  // local variables
  this.engineApi = new EngineApi(
    baseUrl,
    this.options.requestOptions,
    this.options.apiVersion
  );

  if (this.options.pollingDelay) {
    throw new Error('options.pollingDelay got replaced with options.autoPoll');
  }

  // start
  if (this.options.autoPoll) {
    this.start();
  }
}
1198644
protected function writeStatus(Job $job, $status, $type)
    {
        $this->output->writeln(sprintf(
            "<{$type}>[%s][%s] %s</{$type}> %s",
            date('Y-m-d H:i:s'),
            $job->getJobId(),
            str_pad("{$status}:", 11), $job->getName()
        ));
    }
82147
static function __callstatic($func,array $args) {
		if (!self::$fw)
			self::$fw=Base::instance();
		return call_user_func_array(array(self::$fw,$func),$args);
	}
884126
public function th_syncData($node_name, $datas, $type)
    {
        switch ($type) {
            case ClusterProcess::TYPE_UID:
                if (!isset($this->map[$node_name])) {
                    $this->map[$node_name] = new Set($datas);
                } else {
                    $this->map[$node_name]->add(...$datas);
                }
                secho("CLUSTER", "同步$node_name uid信息");
                break;
            case ClusterProcess::TYPE_ACTOR:
                if (!isset($this->actorMap[$node_name])) {
                    $this->actorMap[$node_name] = new Set($datas);
                } else {
                    $this->actorMap[$node_name]->add(...$datas);
                }
                secho("CLUSTER", "同步$node_name actor信息");
                break;
        }
    }
1168138
public function sendByType(int $typeId, string $msg, ?string $dest = null, bool $flush = false) : void
    {
        $msg = $this->composeMsg($msg, $dest);

        $subscriptions = $this->subscriptionTable->findAll()
            ->where('subscription.active', 1)
            ->where('user.active', 1)
            ->where('user:user_subscription_type.subscription_type_id', $typeId);

        foreach ($subscriptions as $row)
        {
            $this->sendNotification($row, $msg, false);
        }

        if ($flush)
        {
            $this->flush();
        }
    }