Multiple Reader

[Jannik-Lappe]

I'm running a nsq server with one topic and four gnsq reader with one channel each on different systems.
When I look at nsqadmin I see two channels with one connection, one channel with five connections and one with six.

Why are there five connections from the same gnsq reader?
Is this normal behavior?

[jehiah]

I suggest you check your process manager; you likely have extra processes running that you don't know about (as a test, try stopping your process and see if all connections are closed, or if only 1 is closed).

[wtolson]

Hi, @Jannik-Lappe. That would not be normal behavior. I suspect @jehiah might be correct. If not however, could you tell me what values you're using to instantiate the gnsq.Reader?

[Jannik-Lappe]

@jehiah
I just restarted the two reader and now they only have one connection O.o
I will monitor this...

@wtolson
Complete NSQ-Reader:

class NsqReader (threading.Thread):
    def __init__(self, name, event, queue, config, channel="default"):
        threading.Thread.__init__(self)
        self.name = name
        self.event = event
        self.config = config
        self.queue = queue
        url = "{}:{}".format(self.config["nsqlookupd"]["name"],
                             self.config["nsqlookupd"]["port"])
        intervall = self.config["nsqlookupd"]["interval"]
        self.reader = gnsq.Reader(message_handler=self.message_handler,
                                  lookupd_http_addresses=url,
                                  lookupd_poll_interval=intervall,
                                  topic=self.config["topics"]["data_topic"],
                                  channel=channel)

    def run(self):
        server = Process(target=self.reader.start)
        try:
            print("Starting " + self.name)
            server.start()
            while not self.event.is_set():
                self.event.wait()
        except Exception as e:
            print("Error! Shutting down\n", e)
        finally:
            self.reader.close()
            self.reader.join(10)
            server.terminate()
            server.join(15)

    def message_handler(self, nsqr, message):
        data = message.body.decode()
        print("receive nsq message")
        self.queue.put(str(data))

I moved the nsq reader to a seperate thread to split it from other services like a mysql-writer.
The process is only used to easily stop the reader and not to block the other threads.

[lambdaq]

I had a similar problem with mixed styles of connections, one is directly connection to nsqd, totally bypass the lookupd, but later we changed to connect via lookupd first. Strange things starts to happen. I had to empty, delete the topic on lookupd and start over for a clean state.

[wtolson]

Closing because of inactivity. Feel free to reopen if you get anymore information.