I believe you all know Publish / Subscribe mode, developers can use third-party open source tools like Redis, NSQ or Nats to implement the subscription mechanism, this article will teach you how to use Go Language to write a stand-alone version of Pub/Sub mode, which is very lightweight in a single system and does not need to rely on third-party services It is easy to implement. The following will directly use a single subscription Topic mechanism to write Publisher and Subscriber.
The first step is to create a Hub to accept multiple Subscribers, and the structure of this Hub structure is as follows.
Initialize subcribers with
newHub, the reason for using map is that it is more convenient to implement unsubscribe later. Next, create the subscriber structure.
The name represents the name of the subscriber, and then the
run function is added to receive messages after a successful subscription.
Receive channel messages by for and select. At the bottom is the initialization of a single Subscriber.
Note that each subscriber receives messages from the Hub through a buffer channel. Please decide whether to adjust the buffer size according to the system context. After initializing the subscriber, you need to drop it into the Hub to subscribe.
Save the subscriber via map and drop it into the background to receive messages via goroutine.
Next, the message is received by the Publisher and dropped to all Subscribers. In the previous step, we saw that the subscriber implements the
run function to accept publisher messages. Let’s see how to implement the publish message mechanism.
The for loop reads out all subscribers and passes in the street message. Next, take a look at how to implement the publish method for a subscriber.
Here we use
select to make sure the whole main will not be blocked. If the message processing is too slow and we don’t use select + default, then the system will be blocked.
If you can subscribe, you need to be able to unsubscribe. That is, how to remove the subscriber from the
map and implement unsubscribe function in the hub.
In addition to
unsubscribe, you can see that we also support the context method to cancel the subscription, so if the developer executes cancel(), theoretically it is also possible to cancel the subscription, and we can modify the
subscribe function here.
Please note that the
go func() listens to the
ctx.Done(), and the cancel() can be executed anywhere in the program to delete the subscriber, and there is a
quit channel in the subscriber structure, which can be used to close the channel after unsubscribe manually, so that the original This allows the original goroutine to end normally and does not cause the system goroutine to keep getting higher.
After completing the above steps, open main.go and start writing the main program.
Verify that the messages come out according to our model. In addition, to verify that all goroutines can be closed properly, use
go.uber.org/goleak to write the test certificate.
Test using context to cancel subscriber.
You can find that in Go language, Pub/Sub mode can be implemented through a simple Buffer Channel, and you can decide whether to import third-party Pub/Sub tools according to the user’s needs. Finally, we attach all code, hope it will be helpful to you.