Learning Golang: Concurrency Pattern Publisher / Subscriber

hi my name is Mario welcome to another learning Go video in today's episode I'm going to be sharing with you another concurrency pattern called publisher subscriber the publisher subscriber pattern is typically used in distributed systems but you can use it also in memory the way I'm going to show you is going to be using go channels and go routines it basically consists of having a publisher that pushes out events or messages and a bunch of subscribers that are interested in receiving those events now a real life example will be when you're looking for maybe some news a specific information about the sports a specific information about different leagues that you like and different teams so you don't want to hear anything or everything about the whole league you are interested in the teams that you like so in this case you are subscribing to specific topics or events that you're interested in it's the same idea so let me show you the code as usual the link to the code that I'm going to be showing you is in the description of this video so feel free to check that out when you you have some time let's init this module you can use whatever module name you want to use I'm just using this one just for to have something what I'm going to be doing first I'm going to be implementing a new BP sub type that is going to be using generics next we need to Define three Fields a slice of channels which indicate the subscribers a Boolean variable that indicates where the pop so is close or not and more importantly a mtic a mutex is used to prevent multiple go routines to access share data then we need to define a method to initialize this type we can use the convention that initialize what we need to initialize in the type internally and go from there next we need three methods one for publishing events one for allowing subscribers to subscribe and one for closing the whole thing thing let's start with the Subscribe method the Subscribe method is going to be doing two things return a new channel that is going to be used for sending new data to this new subscriber and also updating the internal slice of channels that we Define above let me show you in this case what is important is to use the mutex because we are updating an internal field in our type so for doing that you call the mutex with lock and then you just defer the unlock what is this going to be doing is going to be unlocking the lock after the function completes one check that we need to have is that if it's closed we just cannot do anything because the channels that we had before are closed so we just can't continue so in this case we just honor the variable close and we move on now for subscribers we need to have a append that will create a new subscribers and a return field which we didn't Define jet which will be here so we make the channel and we return it so like I said we use the mutex follow by checking if it's close or not we make the new channel and we update it and finally we return it for publish we're going to be doing something similar we implement the function publish but in this case we're going to be using the r lock and again we defer it with SMU R unlock and the reason being is that we are not modifying anything internally in our type we are only reading the values that are available in the subscribers slice again we do a similar check if it's close or not and finally we just use a four to read all the existing subscribers and we send the value to recab we call the mtic we check if it's closed and finally we just push the values out to the subscribers finally the third method will be close again in this case we call the mtic and similar to the Subscribe method we're going to be using lock instead of our lock because we are modifying a field internally which in this case will be the close field and like we did before we check the closed value and we just go and look through all the subscribers and we close each one of them and we update the value of close again we call the mtic we check if it's close or not we close all the existing subscribers and finally we update the varable and we exit now how does this work let me show you the code in Main in main we need to initialize our pop sub so we call new pop up we want to use a string and we're going to be using a weight group so it prevents exiting the main go routine without the other go routines and channels that we're going to be defining for reading and sending values complete so in this case we call it WG now the first thing I'm going to be doing I want to be duplicating this for the sake of just Simplicity but it basically is going to be creating a new subscriber and it's going to be listening to the messages that we're going to be sending for doing that I'm going to using a go routine I'm going to be updating the wait group so it waits and complete after the publisher subscriber type closes let me show you so in this case we just create a typical goutin we add a new value which in the case is a new group we use the four to continue reading messages until there are no more messages we use the select to indicate what are we waiting for we use okay in this case we need to subscribe if it's is nothing we just print the value that says sub one exting meaning that the value of wall here and the value of okay this is value that is coming from the publisher but the value of okay indicates that it's no longer open so we can exit we just say WG done and finally we return and last part will be to print out the value so we can sub one so the subscriber one value F now if we copy and paste all of these and we create a new subscriber so just to differentiate this two so we have S2 here we have S1 subscriber now we have two subscribers both of them will be receiving the same messages now what we need to do next is make sure that the weight group is waiting and start publishing values so the publisher will be pops up publish one two 3 so with that we're going to be publishing three messages the subscribers the two of them will be receiving those message so what is the output of this program let's see before showing the output one thing that I miss will be adding the close and this way will be exiting the gotin cleanly and finally I just want to print out a message that says completed to indicate everything is okay okay let's build the program and run it now if you notice we have the things that we were looking for sub one and sub two are receiving the value of one after that the value of two and after that the value of three and finally it says completed right here because I was using print instead of print L but I guess you get the idea one thing that I want to call out is the way this four is implemented there is a different way to do it I'm prob be a cleaner way to do it do you know how let me know in the comments and that's it this is the concurrency pattern publisher subscriber implemented in go hopefully you find it useful I will talk to you next time until then take care and keep it up see you

Channel: Mario Carrion

Views: 1,927

Keywords: golang, go lang, golang tutorial, go lang tutorial, golang beginners, golang for beginners, learn golang, golang concurrency, golang publisher subscriber, golang publisher, golang subscriber, go tutorial, golang concurrency tutorial, go lang concurrency, go lang concurrency tutorial

Id: s-I3Bs3ZUsY

Length: 8min 16sec (496 seconds)

Published: Mon Nov 06 2023