public function connectToNSQD($host, $port)
    {
        if ($this->isStopped) {
            throw new NsqException("consumer stopped");
        }

        $addr = "$host:$port";
        if (!isset($this->nsqdTCPAddrsConnNum[$addr])) {
            $this->nsqdTCPAddrsConnNum[$addr] = 0;
        }

        $perNsqdMaxNum = $this->getMaxConnNumPerNsqd();
        $currentNum = $this->nsqdTCPAddrsConnNum[$addr];
        $remainNum = max($perNsqdMaxNum - $currentNum, 0);

        $nsqdConns = [];
        for ($i = 0; $i < $remainNum; $i++) {
            try {
                $this->nsqdTCPAddrsConnNum[$addr]++;

                $conn = new Connection($host, $port);
                $hash = spl_object_hash($conn);

                try {
                    $this->pendingConnections[$hash] = $conn;
                    yield $conn->connect();
                    $this->idleConnections[$hash] = $conn;
                } finally {
                    unset($this->pendingConnections[$hash]);
                }

                $nsqdConns[] = $conn;
            } catch (\Throwable $t) {
                $this->nsqdTCPAddrsConnNum[$addr]--;
                throw $t;
            } catch (\Exception $ex) {
                $this->nsqdTCPAddrsConnNum[$addr]--;
                throw $ex;
            }
        }
        yield $nsqdConns;
    }