Метод, который будет вызван,
когда воркер сообщит о новом потоке, который он только что узнал.

@param Worker $worker
@param AbstractThread $thread
@return void
485226
public function threadKnow(Worker $worker, AbstractThread $thread)
    {
        $this->threadKnownIndex->add($thread);
        $worker->addThreadToKnownList($thread);
    }
44641
public function sendToWorkerHandler(MessageInterface $msg)
    {
        if (!$this->isWorkerHandlerReady()) {
            $this->getLogger()->debug('WorkerHandler socket not ready to send.');

            return;
        }

        $this->workerHandlerReady = false;
        $this->workerHandlerStream->send($msg);
    }
1063990
private void addWorker() {
        Throwable ex = null;
        ForkJoinWorkerThread wt = null;
        try {
            if ((wt = factory.newThread(this)) != null) {
                wt.start();
                return;
            }
        } catch (Throwable e) {
            ex = e;
        }
        deregisterWorker(wt, ex); // adjust counts etc on failure
    }
380166
public function setId($id)
    {
        if ($this->status != self::STATUS_NEW) {
            throw new \RuntimeException('Cannot set worker id after worker has started working');
        }

        $this->id = $id;
    }
968067
def _kill_worker_threads(self):
        
        num_workers = len(self._worker_threads)

        if num_workers:
            _log.warning('killing %s active workers(s)', num_workers)
            for worker_ctx, gt in list(self._worker_threads.items()):
                _log.warning('killing active worker for %s', worker_ctx)
                gt.kill()
354008
protected function workInBackground(InputInterface $input): void
    {
        $worker_count = $input->getOption('workers');

        for ($i = 0; $i < $worker_count; ++$i) {
            $pid = pcntl_fork();

            if ($pid === -1) {
                die(sprintf("Could not fork worker %d\n", $i));
            }

            if (!$pid) {
                $pidfile = $input->getOption('pidfile');
                if ($pidfile && $i === 0) {
                    $this->writePidFile($pidfile);
                }

                $worker_factory = $this->worker_factory;

                /** @var array $queues */
                $queues = $input->getArgument('queues');

                /** @var \QueueJitsu\Worker\Worker $worker */
                $worker = $worker_factory($queues);

                $worker($input->getOption('interval'));
                break;
            }
        }
    }
657307
public function reloadWorker($id)
    {
        if (isset($this->workers->threads[$id])) {
            if (!$this->workers->threads[$id]->reloaded) {
                Daemon::$process->log('Spawning worker-replacer for reloaded worker #' . $id);
                $this->spawnWorkers(1);
                $this->workers->threads[$id]->reloaded = true;
            }
        }
    }
715410
private void createThreads(int threadCount)
    {
        
        for(int i=0; i<threadCount; i++) {
            
            AsyncActionWorkerThread worker = new AsyncActionWorkerThread("Async-Action-Worker-"+i);
            this.threadPool.add(worker);
            worker.start();
            
        }
        
    }
1141581
protected function handleGetJobRequest(Worker $worker)
    {
        $this->getLogger()->debug('Worker '.$worker->getHexId().' is requesting a Job.');
        if (!$worker->isValid()) {
            return;
        }
        $actions = $worker->getActionList();
        foreach ($actions as $action) {
            if (isset($this->notAvailable[$action])) {
                unset($this->notAvailable[$action]);
                $this->getLogger()->debug('Action: '.$action.' is available.');
            }
        }
    }
1254769
protected static function forkOneWorkerForLinux($worker)
    {
        // Get available worker id.
        $id = static::getId($worker->workerId, 0);
        if ($id === false) {
            return;
        }
        $pid = pcntl_fork();
        // For master process.
        if ($pid > 0) {
            static::$_pidMap[$worker->workerId][$pid] = $pid;
            static::$_idMap[$worker->workerId][$id]   = $pid;
        } // For child processes.
        elseif (0 === $pid) {
            srand();
            mt_srand();
            if ($worker->reusePort) {
                $worker->listen();
            }
            if (static::$_status === static::STATUS_STARTING) {
                static::resetStd();
            }
            static::$_pidMap  = array();
            // Remove other listener.
            foreach(static::$_workers as $key => $one_worker) {
                if ($one_worker->workerId !== $worker->workerId) {
                    $one_worker->unlisten();
                    unset(static::$_workers[$key]);
                }
            }
            Timer::delAll();
            static::setProcessTitle('WorkerMan: worker process  ' . $worker->name . ' ' . $worker->getSocketName());
            $worker->setUserAndGroup();
            $worker->id = $id;
            $worker->run();
            $err = new Exception('event-loop exited');
            static::log($err);
            exit(250);
        } else {
            throw new Exception("forkOneWorker fail");
        }
    }
1034709
public function working(Worker $worker)
    {
        $this->redis->sadd(self::redisKey(), $this->hostname);

        $this->redis->sadd(self::redisKey($this), (string)$worker);
        $this->redis->expire(self::redisKey($this), $this->timeout);
    }
165117
public function deregisterOnWorkerExit(WorkerInterface $worker)
    {
        $process = $worker->getProcess();
        $process->wait();
        if ($process->isCleanExit()) {
            $this->registry->deregister($worker);
        } else {
            throw new ResqueRuntimeException(
                sprintf(
                    'Foreman error with worker %s wait on pid %d',
                    $worker->getId(),
                    $process->getPid()
                )
            );
        }
    }
928368
public void start() {
		this.workerThread = new Thread(this);
		System.out.println("starting worker thread");
		this.workerThread.start();
		System.out.println("worker thread started");
	}
1055807
private function push(Worker $worker)
    {
        if (!$this->workers->contains($worker)) {
            throw new InvalidArgumentError(
                'The provided worker was not part of this queue.'
            );
        }

        if (0 === ($this->workers[$worker] -= 1)) {
            // Worker is completely idle, remove from busy queue and add to idle queue.
            foreach ($this->busyQueue as $key => $busy) {
                if ($busy === $worker) {
                    unset($this->busyQueue[$key]);
                    break;
                }
            }

            $this->idleWorkers->push($worker);
        }
    }
1005453
public function logPostWorkerEvent(Events\PostWorkerEvent $event)
    {
        $worker = $event->getWorker();

        $this->logger->info(sprintf('Worked an object with "%s"', get_class($worker)));
    }
1092469
protected function handleWorkerInit(Worker $worker, Register $msg)
    {
        $actionCount = intval(count($msg->getActions()));
        if ($actionCount <= 0) {
            $this->getLogger()->error('Invalid init, invalid action count.');

            return;
        }

        foreach ($msg->getActions() as $actionName) {
            $worker->addAction($actionName);

            $action = $this->getActionName($actionName);
            $action->addWorker($worker);
        }

        $this->getLogger()->info('New worker ('.$worker->getHexId().') with '.$actionCount.' actions.');
        $this->getLogger()->debug('Actions for worker '.$worker->getHexId().': '.implode(', ', $worker->getActionList()));

        $worker->setValid(true);

        $this->reply($worker, new HeartbeatResponseWorkerhandler());

        $this->getLogger()->debug('Worker '.bin2hex($worker->getId()).' initialised.');
    }
111940
public function addWorker($worker)
    {
        $this->addDefinition(
            'workers',
            is_callable($worker) ? new CallbackWorker($worker) : $worker
        );

        return $this;
    }