Como construí um sistema Pub/Sub compatível com Redis usando Golang

No último ano, venho construindo , uma alternativa Redis incorporável para o ecossistema Golang. O EchoVault visa replicar a maioria dos recursos do Redis, ao mesmo tempo que fornece uma interface incorporada e uma interface cliente-servidor compatível com clientes Redis existentes que usam o protocolo RESP. o EchoVault Um dos recursos implementados no EchoVault é o recurso Pub/Sub. Este artigo é um breve passo a passo de como o módulo Pub/Sub foi implementado no momento da redação. O que é Pub/Sub? Pub/Sub significa publicar/assinar. Esse padrão permite que os consumidores se inscrevam em determinados canais. Os produtores publicam mensagens nos canais e todos os consumidores inscritos nesse canal recebem a mensagem. Em nosso contexto, devemos habilitar um processo Go para assinar canais específicos quando o EchoVault estiver incorporado. Ao executar no modo cliente-servidor, devemos permitir que uma conexão TCP do cliente se inscreva em canais. No lado da publicação, um processo Go deve ser capaz de publicar em um canal no modo incorporado, e uma conexão cliente TCP deve ser capaz de publicar uma mensagem em um canal no modo cliente-servidor. Requisitos Antes de começarmos, precisamos determinar os requisitos e a pontuação desta implementação. O EchoVault não implementa todos os recursos disponíveis no Redis Pub/Sub no momento em que este artigo foi escrito. No entanto, os recursos principais mais importantes são implementados. Aqui está o que devemos ser capazes de fazer com o EchoVault PubSub: Permita que um cliente TCP se inscreva em uma lista de canais usando o comando . O servidor deve enviar à conexão do cliente uma confirmação das assinaturas. SUBSCRIBE channel [channel …] Uma instância EchoVault incorporada deve ser capaz de se inscrever em uma lista de canais. Permitir que um cliente TCP assine um padrão usando onde o padrão é uma string glob que permite ao cliente receber mensagens publicadas em todos os canais que satisfazem o padrão. PSUBSCRIBE pattern [pattern …] Uma instância EchoVault incorporada deve ser capaz de assinar uma lista de padrões. Permita que um cliente TCP publique uma mensagem em um canal usando o comando . PUBLISH channel message Publique em um canal a partir de uma instância incorporada do EchoVault. Permitir que clientes TCP cancelem a assinatura de canais e padrões usando os comandos e , respectivamente. UNSUBSCRIBE channel [channel …] PUNSUBSCRIBE pattern [pattern …] Permita que a conexão do cliente TCP use o comando para visualizar uma matriz contendo os canais que correspondem ao padrão fornecido. Se nenhum padrão for fornecido, todos os canais ativos serão retornados. Canais ativos são canais com um ou mais assinantes. PUBSUB CHANNELS [pattern] Permitir que clientes TCP usem o comando para visualizar o número de padrões atualmente assinados pelos clientes. PUBSUB NUMPAT Forneça uma API incorporada para visualizar o número de padrões ativos. Permita que clientes TCP usem o comando para visualizar uma matriz de matrizes contendo o nome do canal fornecido e quantos clientes estão atualmente inscritos no canal. PUBSUB NUMSUB [channel [channel ...]] Forneça uma API incorporada para visualizar o número de assinantes em determinados canais. Implementação Agora que temos os requisitos definidos, vamos entrar na implementação. Em um nível superior, teremos uma estrutura PubSub que fornece todas as funcionalidades do PubSub. Teremos também uma estrutura Channel que fornece a funcionalidade para canais e canais padrão. Para esta seção, usaremos o pacote tidwall/resp. Os usuários do EchoVault usam este pacote para enviar respostas RESP aos assinantes do canal. O pacote gobwas/glob lida com a lógica do padrão glob. Canal Primeiro, criaremos a estrutura e todos os seus métodos. Teremos dois tipos de canais: canais regulares e canais padrões. Channel Canais regulares são canais com nomes e sem padrões associados a eles. Os canais padrão usarão o padrão como nome e um padrão glob compilado será associado ao canal. Canais de padrão são usados para assinaturas de padrões. Caso contrário, serão usados canais regulares. A estrutura tem o seguinte formato: 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. } Usaremos o padrão de opção para criar uma nova instância de canal. Aqui estão as duas opções disponíveis: // 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 } O primeiro método para a estrutura Channel é o método . Este método inicia uma goroutine que escuta no as mensagens a serem transmitidas para todos os seus assinantes. Aqui está a implementação: 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() } }() } A goroutine pega a próxima mensagem de , adquire um bloqueio de leitura para os assinantes, transmite a mensagem para cada assinante e então libera o bloqueio de leitura. Como você pode ver, o pacote tidwall/resp nos permite enviar valores RESP para a conexão do cliente facilmente. messageChan A maioria dos clientes Redis espera que o formato da mensagem pub/sub seja uma matriz no seguinte formato: [“mensagem”, , ]. É isso que é transmitido pela EchoVault. Você pode perceber que estamos enviando a mensagem para e não . Estamos usando porque ele fornece métodos auxiliares para gravar valores RESP nele, como . É essencialmente um wrapper em torno , como veremos no método : resp.Conn net.Conn resp.Conn WriteArray net.Conn Subscribe 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 } Junto com o método Subscribe, também precisamos de um método Unsubscribe: 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 } O método Unsubscribe retorna se a conexão foi encontrada e excluída do mapa de assinantes. Caso contrário, ele retorna . Além dos métodos acima, existem mais alguns métodos auxiliares para a estrutura : 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 Usaremos o módulo PubSub para interagir com os canais. A forma da estrutura PubSub é a seguinte: 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{}, } } O primeiro método é o método . Este método trata da inscrição do cliente no(s) canal(is) especificado(s). 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) } } } } } Para inscrever um cliente em um canal, o EchoVault escuta um comando ou . Quando o comando é recebido de um cliente, a conexão TCP do cliente, juntamente com a lista de canais, é passada para o método . Quando um cliente assina usando o comando , o parâmetro será e o valor da variável será definido como “psubscribe”. SUBSCRIBE PSUBSCRIBE Subscribe PSUBSCRIBE withPatterns true action O método permite que um cliente cancele a assinatura de canais ou padrões com base no uso ou . Sua implementação é a seguinte: 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) } Quando o cliente cancela a assinatura de canais/padrões, ele receberá uma resposta contendo uma matriz de matrizes. Cada array interno contém a ação (cancelar assinatura/cancelar assinatura), o nome do canal/padrão e o índice. O próximo método é o método Publish. Este método aceita a mensagem e o nome do canal e então trata da publicação da mensagem internamente. 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 } } } O próximo método é o método , que lida com o comando : Channels PUBSUB CHANNELS [pattern] 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)) } Os 2 métodos a seguir são os métodos e para lidar com e NumPat NumSub PUBSUB NUMPAT Comandos respectivamente. 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) } API incorporada O módulo PubSub espera que uma conexão seja passada para inscrever um cliente em um canal. No entanto, quando o EchoVault está incorporado, não há conexão TCP com um cliente que possa ser usado para uma assinatura. Para contornar isso, usamos para obter as duas extremidades de uma conexão. net.Pipe Uma extremidade da conexão é passada para o manipulador de comandos, que a utiliza para assinar o canal especificado. A outra extremidade da conexão é usada na função retornada. 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) } Aqui está um exemplo de como usar a API SUBSCRIBE e PUBLISH: // 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() Conclusão Este artigo foi um rápido resumo da implementação do Pub/Sub no EchoVault. É claro que há muito contexto em torno do código compartilhado aqui que não cabe em um artigo. Se você estiver interessado no contexto, confira o repositório GitHub do EchoVault e dê uma olhada no que estamos construindo. Se você gosta do que vê, uma estrela do Github seria muito apreciada!