Provide NSQd lookup for producers.
Jump to the last part to understand how to use the test files and bootup nsqds.
Read protocol_v2 documentation (which is right below). protocol_v2 is the recommended protocol.
Two sets of protocols
Recommended:
Protocol v2
Preset producer priority, and connect it to a correpsonding nsqd
API:
//to find a nsqd for a certain prio
producer1, _ := nsq.NewProducer("", config)
err := producer1.ConnectToNSQLookupd_v2(Lookupd_Address, ProducerPriorityLevel:"HIGH"/“LOW”)
err = producer1.Publish(Topic1, MessageBody)
err = producer1.Publish(Topic2, MessageBody)
Explanation:
This protocol links the producer to the nsqd with the required priority.
Once connection is set up, the producer will not switch to another nsqd unless nsqd dies out.
If nsqd dies out, the producer will get a connection error.
This behavior is the original nsq producer behavior.
Advantage:
As long as the user doesn't misbehavior, everything should work fine.
And the relation is neat and beautiful.
However, if there is potential user misbehavior, there will be some performance degradation.
WARNING:
Don't use protocol_v1 and admin_register/admin_unregister with protocol_v2.
Things will become nasty if these two protocols are mingled together.
Set Broadcast-address in the nsqd.sh file to the IP address the nsqd will be serving on.
Protocol v1
Topic oriented protocl. Producer is hooked to nsqd based on the topic priority.
API:
Producer Side
//to find a nsqd for a certain topic
producer1, _ := nsq.NewProducer("", config)
err := producer1.ConnectToNSQLookupd(Lookupd_Address, Topic1)
err = producer1.Publish(Topic1, MessageBody)
Adminstrator Side :
//register or unregister a certain topic in lookupd as a high prio topic
w, _ := nsq.NewProducer("", config)
err := w.RegisterHighPriorityTopic(Lookupd Address, topicName)
err = w.UnregisterHighPriorityTopic(Lookupd Address, topicName)
The Adminstrator is not very dynamic.
Its function is really limited and may cause potential message loss.
It needs to be greatly improved later.
The second and third part of "Some Defects" will talk about it in more detail.
///////////////////////////////////////////////////////////////////////////////////////
Some Big Changes !!
Each producer object can only publish one topic.
So if you want to publish two topics, you need to instance two producers
producer1, _ := nsq.NewProducer("", config)
producer2, _ := nsq.NewProducer("", config)
_ = producer1.ConnectToNSQLookupd(Lookupd_Address, Topic1)
_ = producer2.ConnectToNSQLookupd(Lookupd_Address, Topic2)
err := producer1.Publish(Topic1, MessageBody)
err := producer2.Publish(Topic2, MessageBody)
///////////////////////////////////////////////////////////////////////////////////////
Some Defects (Assume consumers and producers all connect to the lookupd, not nsqds)
1.
Consumers will panic if there are no nsqds currently serving their interested topics.
They dont know which nsqd to connect to. This bug comes with its original code (has nothing to do with our implementation)
So, please run producer first, then consumer. For producers, just send a hello message and sleep for a few seconds to wait for consumer.
2.
When consumer is connected to the lookupd, the change of topic priority might cause messages from a previous producer get lost.
For example:
NSQd1 is responsible for and currently publishing a certain topic topic1.
Then our admin changes the priority of the topic1 during this interval. So NSQd2 is responsible for the topic1.
Later NSQd2 begins to publish the topic.
Then Consumer1 subscribe to the topic1. Yet it can only receive the messages from NSQd2.
The old messages published by NSQd1 are lost.
The reason is NSQd1 has been labeled as not responsible for that topic due to the topic priority change.
NSQd2 is responsible for the topic.
Lookupd will only return the address of NSQd that is currently responsible for the topic to the consumer.
Thats why the message from NSQd1 will be lost.
So to avoid this kind of messages lost, please register or unregister topics when there are no currently ONGOING message transfer of that topic.
3.
While we dynamically register and unregister high prio topics, the active producers will not change their nsqds.
For example:
Producer_1 is publishing high_prio_topic_1 to NSQd_high.
Consumer_1 is subscribing high_prio_topic_1 from NSQd_high.
If we unregister topic_1 (set it to low prio topic), producer_1 is still connecting to NSQd_high even though topic_1 is no longer high prio.
However, for a later publication, producer_2 will correctly connect to the NSQd_low if it is publishing topic_1.
So in other words, the traffic routing is not that dynamic.
Solution:
Uncomment two lines of code in nsqio/go-nsq/producer.go
w.wg.add(1);
go w.lookupdloop();
Explanation:
The client producer library will spawn a go routine that periodically sends http request to lookupd to check which nsqd is currently responsible for the topic.
This method can resolve the discrepency described above.
Yet, I personally dont like it as it is quite brutal and takes a lot of resources.
But to find a better solution, we have to modify the whole NSQlookupd code to make lookupd initiate the connection, which will be really tiring.
///////////////////////////////////////////////////////////////////////////////////////
For nsq folder: make && make install to generate executable for nsqd & nsqlookupd.
In nsq/build, run lookupd.sh to boot up lookupd.
In nsq/build, there are several folders, including nsqd_high, nsqd_low, etc.
Copy the nsqd executable into these folders.
Broadcast-address parameter in nsqd.sh must be set to the ip address the nsqd is listening to.
For NSQ-NSQd-Test folder, the files in there are for test purpose.
Receiver and Sender folders contain producer and consumer
go run receiver.go NUMBER_OF_TOPICS ephemeral/idk
go run sender.go NUMBER_OF_TOPICS ephemeral/idk
All producer and consumers should connect to the same lookupd
Please configure the address of lookupd correctly in both producer/consumer code.
WARNING!!! Run sender.go FIRST, then run receiver.go WARNING END!!!!!!!
The reason is that consumer is going to panic when it realizes that there is no NSQd currently serving the topic.
By making producer sends out a hello msg first, the Lookupd will return the address of the responsible nsqd to the consumer.
The receiver will stop after receiving 1,000 msgs from the producers and output a latency file.
NSQ-NSQd-Test/admin folder
In this folder, there are two go apps.
One is to register high priority topics.
Another is to unregister high priority topics.
go run admin_register.go TOPIC_NAME
go run admin_unregister.go TOPIC_NAME