Golang'ı Kullanarak Redis Uyumlu Bir Pub/Sub Sistemini Nasıl Oluşturdum?

Geçtiğimiz yıl, Golang ekosistemi için yerleştirilebilir bir Redis alternatifi olan inşa ediyordum. EchoVault, hem yerleşik bir arayüz hem de RESP protokolünü kullanan mevcut Redis istemcileriyle uyumlu bir istemci-sunucu arayüzü sağlarken Redis özelliklerinin çoğunu kopyalamayı amaçlamaktadır. EchoVault'u EchoVault'ta uygulanan özelliklerden biri de Pub/Sub özelliğidir. Bu makale, yazıldığı sırada Pub/Sub modülünün nasıl uygulandığına dair kısa bir açıklamadır. Pub/Sub Nedir? Pub/Sub, yayınlama/abone olma anlamına gelir. Bu model tüketicilerin belirli kanallara abone olmalarına olanak tanır. Yapımcılar kanallara mesaj yayınlar ve o kanala abone olan tüm tüketiciler mesajı alır. Bizim bağlamımızda, EchoVault yerleştirildiğinde belirli kanallara abone olmak için bir Go sürecini etkinleştirmeliyiz. İstemci-sunucu modunda çalışırken, istemci TCP bağlantısının kanallara abone olmasına izin vermeliyiz. Yayımlama tarafında, bir Go işlemi, gömülü modda bir kanala yayınlayabilmeli ve bir TCP istemci bağlantısı, istemci-sunucu modunda bir kanala mesaj yayınlayabilmelidir. Gereksinimler Başlamadan önce bu uygulamanın gerekliliklerini ve puanını belirlememiz gerekiyor. EchoVault, bu yazının yazıldığı sırada Redis Pub/Sub'da mevcut olan tüm özellikleri uygulamamaktadır. Ancak en önemli temel özellikler uygulanır. EchoVault PubSub ile şunları yapabilmemiz gerekiyor: Bir TCP istemcisinin komutunu kullanarak bir kanal listesine abone olmasına izin verin. Sunucu, istemci bağlantısına aboneliklerin onayını göndermelidir. SUBSCRIBE channel [channel …] Gömülü bir EchoVault örneği, bir kanal listesine abone olabilmelidir. Bir TCP istemcisinin kullanarak bir kalıba abone olmasına izin verin; burada kalıp, istemcinin kalıbı karşılayan tüm kanallara yayınlanan mesajları almasına olanak tanıyan bir glob dizesidir. PSUBSCRIBE pattern [pattern …] Gömülü bir EchoVault örneği, bir kalıp listesine abone olabilmelidir. Bir TCP istemcisinin, komutunu kullanarak bir kanala mesaj yayınlamasına izin verin. PUBLISH channel message Gömülü bir EchoVault örneğinden bir kanala yayınlayın. Sırasıyla ve komutlarını kullanarak TCP istemcilerinin kanal ve kalıp aboneliğinden çıkmalarına izin verin. UNSUBSCRIBE channel [channel …] PUNSUBSCRIBE pattern [pattern …] Verilen modelle eşleşen kanalları içeren bir diziyi görüntülemek için TCP istemci bağlantısının komutunu kullanmasına izin verin. Herhangi bir model sağlanmazsa tüm aktif kanallar döndürülür. Aktif kanallar bir veya daha fazla abonesi olan kanallardır. PUBSUB CHANNELS [pattern] TCP istemcilerinin, istemciler tarafından halihazırda abone olunan modellerin sayısını görüntülemek için komutunu kullanmasına izin verin. PUBSUB NUMPAT Etkin kalıpların sayısını görüntülemek için yerleşik bir API sağlayın. TCP istemcilerinin, sağlanan kanal adını ve o anda kanala kaç istemcinin abone olduğunu içeren bir dizi diziyi görüntülemek için komutunu kullanmasına izin verin. PUBSUB NUMSUB [channel [channel ...]] Belirli kanallardaki abone sayısını görüntülemek için yerleşik bir API sağlayın. Uygulama Gereksinimleri belirlediğimize göre artık uygulamaya geçebiliriz. Yüksek düzeyde, tüm PubSub işlevlerini sağlayan bir PubSub yapısına sahip olacağız. Ayrıca kanallar ve kalıp kanalları için işlevsellik sağlayan bir Kanal yapısına da sahip olacağız. Bu bölüm için tidwall/resp paketini kullanacağız. EchoVault kullanıcıları bu paketi kanal abonelerine RESP yanıtları göndermek için kullanır. gobwas/glob paketi glob desen mantığını yönetir. Kanal Öncelikle yapısını ve onun tüm metodlarını oluşturacağız. İki tür kanalımız olacak: normal kanallar ve kalıp kanalları. Channel Normal kanallar, adları olan ve bunlarla ilişkilendirilmiş kalıpları olmayan kanallardır. Desen kanalları, modeli ad olarak kullanacak ve derlenmiş bir glob modeli, kanalla ilişkilendirilecektir. Desen kanalları, kalıplara abonelikler için kullanılır. Aksi takdirde normal kanallar kullanılır. yapısı aşağıdaki şekle sahiptir: Channel type Channel struct { name string // Channel name. This can be a glob pattern string. pattern glob.Glob // Compiled glob pattern. This is nil if the channel is not a pattern channel. subscribersRWMut sync.RWMutex // RWMutex to concurrency control when accessing channel subscribers. subscribers map[*net.Conn]*resp.Conn // Map containing the channel subscribers. messageChan *chan string // Messages published to this channel will be sent to this channel. } Yeni bir kanal örneği oluşturmak için seçenek modelini kullanacağız. İşte mevcut iki seçenek: // WithName option sets the channels name. func WithName(name string) func(channel *Channel) { return func(channel *Channel) { channel.name = name } } // WithPattern option sets the compiled glob pattern for the channel if it's a pattern channel. func WithPattern(pattern string) func(channel *Channel) { return func(channel *Channel) { channel.name = pattern channel.pattern = glob.MustCompile(pattern) } } func NewChannel(options ...func(channel *Channel)) *Channel { messageChan := make(chan string, 4096) // messageChan is buffered. This could be a configurable value. channel := &Channel{ name: "", pattern: nil, subscribersRWMut: sync.RWMutex{}, subscribers: make(map[*net.Conn]*resp.Conn), messageChan: &messageChan, } for _, option := range options { option(channel) } return channel } Channel yapısının ilk metodu metodudur. Bu yöntem, tüm abonelerine yayınlanacak mesajları dinleyen bir goroutine başlatır. İşte uygulama: Start messageChan func (ch *Channel) Start() { go func() { for { message := <-*ch.messageChan ch.subscribersRWMut.RLock() for _, conn := range ch.subscribers { go func(conn *resp.Conn) { if err := conn.WriteArray([]resp.Value{ resp.StringValue("message"), resp.StringValue(ch.name), resp.StringValue(message), }); err != nil { log.Println(err) } }(conn) } ch.subscribersRWMut.RUnlock() } }() } Goroutine, bir sonraki mesajı alır, aboneler için bir okuma kilidi elde eder, mesajı her aboneye yayınlar ve ardından okuma kilidini serbest bırakır. Gördüğünüz gibi tidwall/resp paketi RESP değerlerini client bağlantısına kolaylıkla göndermemizi sağlıyor. messageChan Redis istemcilerinin çoğu, pub/sub mesaj formatının şu şekilde bir dizi olmasını bekler: [“mesaj”, , ]. EchoVault tarafından yayınlanan şey budur. Mesajı değil gönderdiğimizi fark edebilirsiniz. RESP değerlerini yazmak için gibi yardımcı yöntemler sağladığı için kullanıyoruz. yönteminde göreceğimiz gibi, aslında çevreleyen bir sarmalayıcıdır: net.Conn resp.Conn WriteArray resp.Conn Subscribe net.Conn func (ch *Channel) Subscribe(conn *net.Conn) bool { ch.subscribersRWMut.Lock() // Acquire write-lock because we'll be modifying the subscriber map. defer ch.subscribersRWMut.Unlock() // If the connection does not exist in the subscriber map, add it. if _, ok := ch.subscribers[conn]; !ok { ch.subscribers[conn] = resp.NewConn(*conn) } _, ok := ch.subscribers[conn] return ok } Abone Ol yönteminin yanı sıra Abonelikten Çık yöntemine de ihtiyacımız var: func (ch *Channel) Unsubscribe(conn *net.Conn) bool { ch.subscribersRWMut.Lock() defer ch.subscribersRWMut.Unlock() if _, ok := ch.subscribers[conn]; !ok { return false } delete(ch.subscribers, conn) return true } Bağlantının bulunması ve abone haritasından silinmesi durumunda Abonelikten Çıkma yöntemi döndürür. Aksi takdirde döndürür. Yukarıdaki yöntemlere ek olarak yapısı için birkaç yardımcı yöntem daha vardır: true false Channel func (ch *Channel) Name() string { return ch.name } func (ch *Channel) Pattern() glob.Glob { return ch.pattern } func (ch *Channel) Publish(message string) { *ch.messageChan 0 return active } // NumSubs returns the number of subscribers for this channel. func (ch *Channel) NumSubs() int { ch.subscribersRWMut.RLock() defer ch.subscribersRWMut.RUnlock() n := len(ch.subscribers) return n } // Subscribers returns a copy of the subscriber map. func (ch *Channel) Subscribers() map[*net.Conn]*resp.Conn { ch.subscribersRWMut.RLock() defer ch.subscribersRWMut.RUnlock() subscribers := make(map[*net.Conn]*resp.Conn, len(ch.subscribers)) for k, v := range ch.subscribers { subscribers[k] = v } return subscribers } PubSub Kanallarla etkileşim kurmak için PubSub modülünü kullanacağız. PubSub yapısının şekli aşağıdaki gibidir: type PubSub struct { channels []*Channel // Slice of references to channels channelsRWMut sync.RWMutex // RWMutex for concurrency controls when accessing channels } func NewPubSub() *PubSub { return &PubSub{ channels: []*Channel{}, channelsRWMut: sync.RWMutex{}, } } İlk yöntem yöntemidir. Bu yöntem, istemcinin belirtilen kanal(lar)a abone olmasını sağlar. Subscribe func (ps *PubSub) Subscribe(_ context.Context, conn *net.Conn, channels []string, withPattern bool) { ps.channelsRWMut.Lock() // Acquire write-lock as we may edit the slice of channels. defer ps.channelsRWMut.Unlock() r := resp.NewConn(*conn) // Wrap net.Conn connection with resp.Conn. action := "subscribe" if withPattern { action = "psubscribe" } // Loop through all the channels that the client has requested to subscribe to. for i := 0; i < len(channels); i++ { // Check if channel with given name exists // If it does, subscribe the connection to the channel // If it does not, create the channel and subscribe to it channelIdx := slices.IndexFunc(ps.channels, func(channel *Channel) bool { return channel.name == channels[i] }) if channelIdx == -1 { // If the channel does not exist, create new channel, start it, and subscribe to it. var newChan *Channel if withPattern { newChan = NewChannel(WithPattern(channels[i])) } else { newChan = NewChannel(WithName(channels[i])) } newChan.Start() if newChan.Subscribe(conn) { // Write string array to the client connection confirming the subscription. if err := r.WriteArray([]resp.Value{ resp.StringValue(action), resp.StringValue(newChan.name), resp.IntegerValue(i + 1), }); err != nil { log.Println(err) } } ps.channels = append(ps.channels, newChan) // Append the new channel to the list of channels. } else { // Subscribe to existing channel if ps.channels[channelIdx].Subscribe(conn) { // Write string array to the client connection confirming the subscription. if err := r.WriteArray([]resp.Value{ resp.StringValue(action), resp.StringValue(ps.channels[channelIdx].name), resp.IntegerValue(i + 1), }); err != nil { log.Println(err) } } } } } Bir istemciyi bir kanala abone olmak için EchoVault, veya komutunu dinler. Komut bir istemciden alındığında, istemcinin TCP bağlantısı kanal listesiyle birlikte yöntemine iletilir. Bir istemci komutunu kullanarak abone olduğunda parametresi olacak ve değişkeninin değeri "psubscribe" olarak ayarlanacaktır. SUBSCRIBE PSUBSCRIBE Subscribe PSUBSCRIBE withPatterns true action yöntemi, bir müşterinin veya kullanıp kullanmadığına bağlı olarak kanallardan veya kalıplardan aboneliğini iptal etmesine olanak tanır. Uygulaması şu şekildedir: Unsubscribe SUBSCRIBE PSUBSCRIBE func (ps *PubSub) Unsubscribe(_ context.Context, conn *net.Conn, channels []string, withPattern bool) []byte { ps.channelsRWMut.RLock() defer ps.channelsRWMut.RUnlock() action := "unsubscribe" if withPattern { action = "punsubscribe" } unsubscribed := make(map[int]string) // A map of all the channels/patterns successfully unsubscribed from. idx := 1 // idx holds the 1-based index of the channel/pattern unsubscribed from. if len(channels) <= 0 { if !withPattern { // If the channels slice is empty, and no pattern is provided // unsubscribe from all channels. for _, channel := range ps.channels { if channel.pattern != nil { // Skip pattern channels continue } if channel.Unsubscribe(conn) { unsubscribed[idx] = channel.name idx += 1 } } } else { // If the channels slice is empty, and pattern is provided // unsubscribe from all patterns. for _, channel := range ps.channels { if channel.pattern == nil { // Skip non-pattern channels continue } if channel.Unsubscribe(conn) { unsubscribed[idx] = channel.name idx += 1 } } } } // Unsubscribe from channels where the name exactly matches channel name. // If unsubscribing from a pattern, also unsubscribe from all channel whose // names exactly matches the pattern name. for _, channel := range ps.channels { // For each channel in PubSub for _, c := range channels { // For each channel name provided if channel.name == c && channel.Unsubscribe(conn) { unsubscribed[idx] = channel.name idx += 1 } } } // If withPattern is true, unsubscribe from channels where pattern matches pattern provided, // also unsubscribe from channels where the name matches the given pattern. if withPattern { for _, pattern := range channels { g := glob.MustCompile(pattern) for _, channel := range ps.channels { // If it's a pattern channel, directly compare the patterns if channel.pattern != nil && channel.name == pattern { if channel.Unsubscribe(conn) { unsubscribed[idx] = channel.name idx += 1 } continue } // If this is a regular channel, check if the channel name matches the pattern given if g.Match(channel.name) { if channel.Unsubscribe(conn) { unsubscribed[idx] = channel.name idx += 1 } } } } } // Construct a RESP response confirming the channels/patterns unsubscribed from. res := fmt.Sprintf("*%d\r\n", len(unsubscribed)) for key, value := range unsubscribed { res += fmt.Sprintf("*3\r\n+%s\r\n$%d\r\n%s\r\n:%d\r\n", action, len(value), value, key) } return []byte(res) } İstemci kanal/kalıp aboneliğinden çıktığında, bir dizi diziyi içeren bir yanıt alacaktır. Her bir iç dizi, eylemi (abonelikten çıkma/aboneliği iptal etme), kanal/kalıp adını ve dizini içerir. Bir sonraki yöntem Yayınla yöntemidir. Bu yöntem, mesajı ve kanal adını kabul eder ve ardından mesajın dahili olarak yayınlanmasını gerçekleştirir. func (ps *PubSub) Publish(_ context.Context, message string, channelName string) { ps.channelsRWMut.RLock() defer ps.channelsRWMut.RUnlock() // Loop through all of the existing channels. for _, channel := range ps.channels { // If it's a regular channel, check if the channel name matches the name given. if channel.pattern == nil { if channel.name == channelName { channel.Publish(message) // Publish the message to the channel. } continue } // If it's a glob pattern channel, check if the provided channel name matches the pattern. if channel.pattern.Match(channelName) { channel.Publish(message) // Publish the message to the channel } } } Sonraki yöntem, komutunu işleyen yöntemidir: PUBSUB CHANNELS [pattern] Channels func (ps *PubSub) Channels(pattern string) []byte { ps.channelsRWMut.RLock() defer ps.channelsRWMut.RUnlock() var count int var res string // If pattern is an empty string, return all the active channels. if pattern == "" { for _, channel := range ps.channels { if channel.IsActive() { res += fmt.Sprintf("$%d\r\n%s\r\n", len(channel.name), channel.name) count += 1 } } res = fmt.Sprintf("*%d\r\n%s", count, res) return []byte(res) } g := glob.MustCompile(pattern) for _, channel := range ps.channels { // If channel is a pattern channel, then directly compare the channel name to pattern. if channel.pattern != nil && channel.name == pattern && channel.IsActive() { res += fmt.Sprintf("$%d\r\n%s\r\n", len(channel.name), channel.name) count += 1 continue } // Channel is not a pattern channel. Check if the channel name matches the provided glob pattern. if g.Match(channel.name) && channel.IsActive() { res += fmt.Sprintf("$%d\r\n%s\r\n", len(channel.name), channel.name) count += 1 } } // Return a RESP array containing all the active channel names. return []byte(fmt.Sprintf("*%d\r\n%s", count, res)) } Aşağıdaki 2 yöntem işlemek için ve yöntemleridir ve PUBSUB NUMPAT NumPat NumSub komutları sırasıyla. PUBSUB NUMSUB [channel [channel …]] func (ps *PubSub) NumPat() int { ps.channelsRWMut.RLock() defer ps.channelsRWMut.RUnlock() var count int for _, channel := range ps.channels { if channel.pattern != nil && channel.IsActive() { count += 1 } } return count } func (ps *PubSub) NumSub(channels []string) []byte { ps.channelsRWMut.RLock() defer ps.channelsRWMut.RUnlock() res := fmt.Sprintf("*%d\r\n", len(channels)) for _, channel := range channels { // If it's a pattern channel, skip it chanIdx := slices.IndexFunc(ps.channels, func(c *Channel) bool { return c.name == channel }) if chanIdx == -1 { res += fmt.Sprintf("*2\r\n$%d\r\n%s\r\n:0\r\n", len(channel), channel) continue } res += fmt.Sprintf("*2\r\n$%d\r\n%s\r\n:%d\r\n", len(channel), channel, ps.channels[chanIdx].NumSubs()) } return []byte(res) } Gömülü API PubSub modülü, bir istemcinin bir kanala abone olması için bir bağlantının iletilmesini bekler. Ancak EchoVault gömülü olduğunda istemciye abonelik için kullanılabilecek bir TCP bağlantısı yoktur. Bunu aşmak için bağlantının her iki ucunu da almak amacıyla kullanırız. net.Pipe Bağlantının bir ucu, onu belirtilen kanala abone olmak için kullanan komut işleyicisine iletilir. Bağlantının diğer ucu, döndürülen işlevinde kullanılır. ReadPubSubMessage type conn struct { readConn *net.Conn writeConn *net.Conn } var connections map[string]conn // ReadPubSubMessage is returned by the SUBSCRIBE and PSUBSCRIBE functions. // // This function is lazy, therefore it needs to be invoked in order to read the next message. // When the message is read, the function returns a string slice with 3 elements. // Index 0 holds the event type which in this case will be "message". Index 1 holds the channel name. // Index 2 holds the actual message. type ReadPubSubMessage func() []string // SUBSCRIBE subscribes the caller to the list of provided channels. // // Parameters: // // `tag` - string - The tag used to identify this subscription instance. // // `channels` - ...string - The list of channels to subscribe to. // // Returns: ReadPubSubMessage function which reads the next message sent to the subscription instance. // This function is blocking. func (server *EchoVault) SUBSCRIBE(tag string, channels ...string) ReadPubSubMessage { // Initialize connection tracker if calling subscribe for the first time if connections == nil { connections = make(map[string]conn) } // If connection with this name does not exist, create new connection it var readConn net.Conn var writeConn net.Conn if _, ok := connections[tag]; !ok { readConn, writeConn = net.Pipe() connections[tag] = conn{ readConn: &readConn, writeConn: &writeConn, } } // Subscribe connection to the provided channels cmd := append([]string{"SUBSCRIBE"}, channels...) go func() { _, _ = server.handleCommand(server.context, internal.EncodeCommand(cmd), connections[tag].writeConn, false) }() return func() []string { r := resp.NewConn(readConn) v, _, _ := r.ReadValue() res := make([]string, len(v.Array())) for i := 0; i < len(res); i++ { res[i] = v.Array()[i].String() } return res } } // UNSUBSCRIBE unsubscribes the caller from the given channels. // // Parameters: // // `tag` - string - The tag used to identify this subscription instance. // // `channels` - ...string - The list of channels to unsubscribe from. func (server *EchoVault) UNSUBSCRIBE(tag string, channels ...string) { if connections == nil { return } if _, ok := connections[tag]; !ok { return } cmd := append([]string{"UNSUBSCRIBE"}, channels...) _, _ = server.handleCommand(server.context, internal.EncodeCommand(cmd), connections[tag].writeConn, false) } // PSUBSCRIBE subscribes the caller to the list of provided glob patterns. // // Parameters: // // `tag` - string - The tag used to identify this subscription instance. // // `patterns` - ...string - The list of glob patterns to subscribe to. // // Returns: ReadPubSubMessage function which reads the next message sent to the subscription instance. // This function is blocking. func (server *EchoVault) PSUBSCRIBE(tag string, patterns ...string) ReadPubSubMessage { // Initialize connection tracker if calling subscribe for the first time if connections == nil { connections = make(map[string]conn) } // If connection with this name does not exist, create new connection it var readConn net.Conn var writeConn net.Conn if _, ok := connections[tag]; !ok { readConn, writeConn = net.Pipe() connections[tag] = conn{ readConn: &readConn, writeConn: &writeConn, } } // Subscribe connection to the provided channels cmd := append([]string{"PSUBSCRIBE"}, patterns...) go func() { _, _ = server.handleCommand(server.context, internal.EncodeCommand(cmd), connections[tag].writeConn, false) }() return func() []string { r := resp.NewConn(readConn) v, _, _ := r.ReadValue() res := make([]string, len(v.Array())) for i := 0; i < len(res); i++ { res[i] = v.Array()[i].String() } return res } } // PUNSUBSCRIBE unsubscribes the caller from the given glob patterns. // // Parameters: // // `tag` - string - The tag used to identify this subscription instance. // // `patterns` - ...string - The list of glob patterns to unsubscribe from. func (server *EchoVault) PUNSUBSCRIBE(tag string, patterns ...string) { if connections == nil { return } if _, ok := connections[tag]; !ok { return } cmd := append([]string{"PUNSUBSCRIBE"}, patterns...) _, _ = server.handleCommand(server.context, internal.EncodeCommand(cmd), connections[tag].writeConn, false) } ABONE OL ve YAYINLA API'sinin nasıl kullanılacağına dair bir örnek: // Subscribe to multiple EchoVault channels. readMessage := server.SUBSCRIBE("subscriber1", "channel_1", "channel_2", "channel_3") wg.Add(1) go func() { wg.Done() for { message := readMessage() fmt.Printf("EVENT: %s, CHANNEL: %s, MESSAGE: %s\n", message[0], message[1], message[2]) } }() wg.Wait() wg.Add(1) go func() { for i := 1; i <= 3; i++ { // Simulating delay. <-time.After(1 * time.Second) // Publish message to each EchoVault channel. _, _ = server.PUBLISH(fmt.Sprintf("channel_%d", i), "Hello!") } wg.Done() }() wg.Wait() Çözüm Bu makale, EchoVault'taki Pub/Sub uygulamasının kısa bir özetiydi. Elbette, burada paylaşılan kodun etrafında tek bir makaleye sığmayacak kadar çok bağlam var. Bağlamla ilgileniyorsanız EchoVault'un GitHub deposuna göz atın ve ne oluşturduğumuza bir göz atın. Gördüğünüzü beğendiyseniz Github yıldızı çok takdir edilecektir!