GoSudaConfiguração de servidor NATS embarcado em aplicações Go e estrutura de comunicação em processo
By prravda views ...
Primeiros passos
Sobre o NATS
Aplicações e serviços de software precisam trocar dados. O NATS é uma infraestrutura que permite essa troca de dados, segmentada na forma de mensagens. Chamamos isso de "message oriented middleware" (middleware orientado a mensagens).
Com o NATS, desenvolvedores de aplicações podem:
- Construir aplicações cliente-servidor distribuídas e escaláveis de forma simples.
- Armazenar e distribuir dados em tempo real de maneira geral. Isso pode ser alcançado de forma flexível em diversos ambientes, linguagens, provedores de nuvem e sistemas on-premises.
- O NATS é um message broker construído em Go.
NATS incorporado
Se sua aplicação está em Go, e se isso se adequa ao seu caso de uso e aos cenários de implantação, você pode até incorporar um servidor NATS dentro da sua aplicação.
- Há também uma característica peculiar do NATS: no caso de aplicações construídas em Go, ele oferece suporte ao embedded mode.
- Ou seja, em vez da forma comum dos message brokers, que consiste em iniciar um servidor de broker separado e comunicar-se com ele por meio dos clientes da aplicação, é possível embutir o próprio broker (embed) na aplicação desenvolvida em Go.
Benefícios e casos de uso do NATS incorporado
- Como existe um vídeo do Youtube bem elaborado sobre o tema, o link do vídeo substitui uma explicação adicional.
- Mesmo sem implantar um servidor de message broker separado, é possível criar uma modular monolith application e alcançar separate of concern, aproveitando a vantagem de embutir o NATS como embedded. Além disso, torna-se possível a single binary deployment.
- Ele pode ser usado de forma útil não apenas em platform with no network (wasm), mas também em offline-first application.
Exemplo da documentação oficial
1package main
2
3import (
4 "fmt"
5 "time"
6
7 "github.com/nats-io/nats-server/v2/server"
8 "github.com/nats-io/nats.go"
9)
10
11func main() {
12 // Opções do servidor
13 opts := &server.Options{}
14
15 // Inicializa um novo servidor com as opções
16 ns, err := server.NewServer(opts)
17
18 if err != nil {
19 panic(err)
20 }
21
22 // Inicia o servidor via goroutine
23 go ns.Start()
24
25 // Aguarda até que o servidor esteja pronto para conexões
26 if !ns.ReadyForConnections(4 * time.Second) {
27 panic("not ready for connection")
28 }
29
30 // Conecta ao servidor
31 nc, err := nats.Connect(ns.ClientURL())
32
33 if err != nil {
34 panic(err)
35 }
36
37 subject := "my-subject"
38
39 // Assina o subject
40 nc.Subscribe(subject, func(msg *nats.Msg) {
41 // Imprime os dados da mensagem
42 data := string(msg.Data)
43 fmt.Println(data)
44
45 // Desliga o servidor (opcional)
46 ns.Shutdown()
47 })
48
49 // Publica dados no subject
50 nc.Publish(subject, []byte("Hello embedded NATS!"))
51
52 // Aguarda o desligamento do servidor
53 ns.WaitForShutdown()
54}
- Este é o exemplo de Embedded NATS apresentado na documentação oficial do NATS, mas, seguindo o código de exemplo como está, a comunicação não ocorre no embedding mode.
1Every 2.0s: netstat -an | grep 127.0.0.1 pravdalaptop-home.local: 02:34:20
2 in 0.017s (0)
3...
4tcp4 0 0 127.0.0.1.4222 127.0.0.1.63769 TIME_WAIT
- Ao executar o arquivo go com
go run .enquanto se monitoram, pelo comandowatch 'netstat -an | grep 127.0.0.1', as redes que trafegam por localhost(127.0.0.1), observa-se que surgem novas requisições de rede originadas na porta padrão do NATS,4222.
Configurações corretas para o embedding mode
Para se comunicar no embedded mode conforme pretendido, são necessárias duas opções como segue.
- Client: é preciso incluir a opção
InProcessServer. - Server: é preciso declarar explicitamente a flag
DontListencomotrueemServer.Options.
- Client: é preciso incluir a opção
Essas partes não estão oficialmente documentadas, e o início dessa funcionalidade pode ser identificado por meio deste PR.
Este PR adiciona três coisas:
- A função
InProcessConn()emServer, que constrói umnet.Pipepara obter uma conexão com o servidor NATS sem usar sockets TCP - A opção
DontListen, que instrui o servidor NATS a não escutar no listener TCP habitual - Um canal
startupComplete, que é fechado logo antes de iniciarmos oAcceptLoop, ereadyForConnectionsaguardará por ele
A principal motivação é que temos uma aplicação que pode executar tanto em modo monolith (processo único) quanto em modo polylith (múltiplos processos). Gostaríamos de usar o NATS nos dois modos por simplicidade, mas o modo monolith precisa atender a diversas plataformas onde abrir conexões por socket não faz sentido (mobile) ou simplesmente não é possível (WASM). Essas mudanças permitirão usar o NATS inteiramente em processo.
Um PR complementar nats-io/nats.go#774 adiciona suporte no lado do cliente.
Este é meu primeiro PR para este projeto, então peço desculpas antecipadamente caso eu tenha deixado passar algo óbvio.
/cc @nats-io/core
- A função
Exemplo funcional para o embedded mode
1package main
2
3import (
4 "fmt"
5 "time"
6
7 "github.com/nats-io/nats-server/v2/server"
8 "github.com/nats-io/nats.go"
9)
10
11func main() {
12 opts := &server.Options{
13 // Para configurar o servidor NATS embutido
14 // defina DonListen como true
15 DontListen: true,
16 }
17
18 // Inicializa um novo servidor com as opções
19 ns, err := server.NewServer(opts)
20
21 if err != nil {
22 panic(err)
23 }
24
25 // Inicia o servidor via goroutine
26 go ns.Start()
27
28 // Aguarda até que o servidor esteja pronto para conexões
29 if !ns.ReadyForConnections(10 * time.Second) {
30 panic("not ready for connection")
31 }
32
33 // Conecta ao servidor via conexão in-process
34 nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
35
36 if err != nil {
37 panic(err)
38 }
39
40 subject := "my-subject"
41
42 // Assina o subject
43 nc.Subscribe(subject, func(msg *nats.Msg) {
44 // Imprime os dados da mensagem
45 data := string(msg.Data)
46 fmt.Println(data)
47
48 // Desliga o servidor (opcional)
49 ns.Shutdown()
50 })
51
52 // Publica dados no subject
53 nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55 // Aguarda o desligamento do servidor
56 ns.WaitForShutdown()
57}
58
1Every 2.0s: netstat -an | grep 127.0.0.1 pravdalaptop-home.local: 02:37:50
2 in 0.023s (0)
3...no additional logs
4
- Agora é possível observar que nenhum hop de rede adicional ocorre, conforme pretendido.
Por dentro do funcionamento
TL;DR
- Este é o diagrama de sequência que mostra quais funções são executadas internamente e como funcionam quando o código é executado a partir de
main.go. Em linhas gerais, os pontos principais são os seguintes.- Com
DontListen: true, o servidor omite a fase de escuta de clientes denominadaAcceptLoop. - Se a opção
InProcessServerdo Connect do client estiver ativada, é criada uma conexão em memória, um pipe é construído vianet.Pipee a extremidade do pipe é retornada ao client como um tiponet.Conn. - Client e server realizam a comunicação in-process por essa conexão.
- Com
Servidor
AcceptLoop
1// nats-server/server/server.go
2
3// Aguarda os clientes.
4if !opts.DontListen {
5 s.AcceptLoop(clientListenReady)
6}
- Primeiramente, quando
DontListené true, a fase de escuta de clientes denominadaAcceptLoopé omitida.
1// nats-server/server/server.go
2
3// AcceptLoop é exportada para facilitar os testes.
4func (s *Server) AcceptLoop(clr chan struct{}) {
5 // Se saíssemos antes de o listener ser configurado corretamente,
6 // garanta que o canal seja fechado.
7 defer func() {
8 if clr != nil {
9 close(clr)
10 }
11 }()
12
13 if s.isShuttingDown() {
14 return
15 }
16
17 // Captura um snapshot das opções do servidor.
18 opts := s.getOpts()
19
20 // Configura o estado que permite o shutdown
21 s.mu.Lock()
22 hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
23 l, e := natsListen("tcp", hp)
24 s.listenerErr = e
25 if e != nil {
26 s.mu.Unlock()
27 s.Fatalf("Error listening on port: %s, %q", hp, e)
28 return
29 }
30 s.Noticef("Listening for client connections on %s",
31 net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))
32
33 // Alerta que TLS está habilitado.
34 if opts.TLSConfig != nil {
35 s.Noticef("TLS required for client connections")
36 if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
37 s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
38 }
39 }
40
41 // Se o servidor foi iniciado com RANDOM_PORT (-1), opts.Port seria igual
42 // a 0 no início desta função. Por isso, precisamos obter a porta real
43 if opts.Port == 0 {
44 // Grava a porta resolvida de volta nas opções.
45 opts.Port = l.Addr().(*net.TCPAddr).Port
46 }
47
48 // Agora que a porta foi definida (caso tenha sido RANDOM), configura o
49 // Host/Port de INFO do servidor com os valores de Options ou
50 // ClientAdvertise.
51 if err := s.setInfoHostPort(); err != nil {
52 s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
53 l.Close()
54 s.mu.Unlock()
55 return
56 }
57 // Mantém o registro das URLs de conexão dos clientes, que podem ser necessárias depois.
58 s.clientConnectURLs = s.getClientConnectURLs()
59 s.listener = l
60
61 go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
62 func(_ error) bool {
63 if s.isLameDuckMode() {
64 // Sinaliza que não estamos aceitando novos clientes
65 s.ldmCh <- true
66 // Agora aguarda o Shutdown...
67 <-s.quitCh
68 return true
69 }
70 return false
71 })
72 s.mu.Unlock()
73
74 // Informa ao chamador que estamos prontos
75 close(clr)
76 clr = nil
77}
- Como referência, a função AcceptLoop executa o seguinte processo. Trata-se de partes relacionadas à comunicação em rede, como
TLSehostPort, que são desnecessárias na comunicação in-process e podem ser omitidas sem prejuízo.
Client
InProcessServer
1
2// nats-go/nats.go
3
4// Connect tentará conectar-se ao sistema NATS.
5// A url pode conter semântica de username/password. por exemplo nats://derek:pass@localhost:4222
6// Também são suportados arrays separados por vírgula, por exemplo urlA, urlB.
7// As opções começam com os padrões, mas podem ser sobrescritas.
8// Para conectar-se à porta websocket de um servidor NATS, use o esquema `ws` ou `wss`, como
9// `ws://localhost:8080`. Observe que esquemas websocket não podem ser misturados com outros (nats/tls).
10func Connect(url string, options ...Option) (*Conn, error) {
11 opts := GetDefaultOptions()
12 opts.Servers = processUrlString(url)
13 for _, opt := range options {
14 if opt != nil {
15 if err := opt(&opts); err != nil {
16 return nil, err
17 }
18 }
19 }
20 return opts.Connect()
21}
1// nats-go/nats.go
2
3// Options pode ser usada para criar uma conexão personalizada.
4type Options struct {
5 // Url representa uma única URL de servidor NATS à qual o client
6 // se conectará. Se a opção Servers também estiver definida, ela
7 // passa a ser o primeiro servidor do array Servers.
8 Url string
9
10 // InProcessServer representa um servidor NATS em execução dentro do
11 // mesmo processo. Se estiver definida, tentaremos conectar-nos ao
12 // servidor diretamente, em vez de usar conexões TCP externas.
13 InProcessServer InProcessConnProvider
14
15 //...
16}
1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4 InProcessConn() (net.Conn, error)
5}
- A função
Connect, que realiza a conexão entre o servidor nats e o client nats, recebe a URL do client e as opções de conexão, e a struct Options que reúne essas opções possui um campo chamadoInProcessServerdo tipo interfaceInProcesConnProvider.
1// main.go of example code
2
3// Inicializa um novo servidor com as opções
4ns, err := server.NewServer(opts)
5
6//...
7
8// Conecta ao servidor via conexão in-process
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
- Quando o client nats executa Connect e passa
nats.InProcessServer(ns)para o campoInProcessServer,
1// nats-go/nats.go
2
3// InProcessServer é uma Option que tenta estabelecer uma conexão com um servidor NATS
4// em execução dentro do processo, em vez de discar via TCP.
5func InProcessServer(server InProcessConnProvider) Option {
6 return func(o *Options) error {
7 o.InProcessServer = server
8 return nil
9 }
10}
- o InProcessServer da option é substituído pelo servidor NATS embutido,
1// nats-go/nats.go
2
3// createConn conecta-se ao servidor e envolve as estruturas bufio
4// apropriadas. Ele faz a coisa certa quando já existe uma
5// conexão estabelecida.
6func (nc *Conn) createConn() (err error) {
7 if nc.Opts.Timeout < 0 {
8 return ErrBadTimeout
9 }
10 if _, cur := nc.currentServer(); cur == nil {
11 return ErrNoServers
12 }
13
14 // Se temos uma referência a um servidor in-process, estabelecemos uma
15 // conexão usando-a.
16 if nc.Opts.InProcessServer != nil {
17 conn, err := nc.Opts.InProcessServer.InProcessConn()
18 if err != nil {
19 return fmt.Errorf("failed to get in-process connection: %w", err)
20 }
21 nc.conn = conn
22 nc.bindToNewConn()
23 return nil
24 }
25
26 //...
27}
- Essa interface é usada na função
createConn, que cria a conexão. Quando a optionInProcessServernão é nil (ou seja, é válida), ela executa oInProcessConndo InProcessServer presente na option,
1// nats-server/server/server.go
2
3// InProcessConn retorna uma conexão in-process com o servidor,
4// evitando a necessidade de um listener TCP para conectividade local
5// dentro do mesmo processo. Pode ser usada independentemente do
6// estado da opção DontListen.
7func (s *Server) InProcessConn() (net.Conn, error) {
8 pl, pr := net.Pipe()
9 if !s.startGoRoutine(func() {
10 s.createClientInProcess(pl)
11 s.grWG.Done()
12 }) {
13 pl.Close()
14 pr.Close()
15 return nil, fmt.Errorf("failed to create connection")
16 }
17 return pr, nil
18}
- chamando o
InProcessConnimplementado no server. - Essa função é chamada pelo
nats.go, o client em Go do NATS, quando oInProcessServerdo nc (conexão nats) não é nil; ela cria a conexão (net.Conn) e a vincula à conexão do server.
Interface de Go definida pelo consumidor
Um tipo implementa uma interface simplesmente implementando seus métodos. Não há declaração explícita de intenção, nem palavra-chave "implements". Interfaces implícitas desacoplam a definição de uma interface de sua implementação, que pode então aparecer em qualquer pacote sem combinação prévia.
Interfaces are implemented implicitly, A Tour of Go
Se um tipo existe apenas para implementar uma interface e nunca terá métodos exportados além dessa interface, não há necessidade de exportar o próprio tipo.
- Esse design de interface incorpora bem a chamada consumer defined interface e o structural typing (duck typing) que se costuma mencionar em Go, por isso gostaria de apresentar também esse tema.
1// nats-go/nats.go
2
3// Options pode ser usada para criar uma conexão personalizada.
4type Options struct {
5 // Url representa uma única URL de servidor NATS à qual o client
6 // se conectará. Se a opção Servers também estiver definida, ela
7 // passa a ser o primeiro servidor do array Servers.
8 Url string
9
10 // InProcessServer representa um servidor NATS em execução dentro do
11 // mesmo processo. Se estiver definida, tentaremos conectar-nos ao
12 // servidor diretamente, em vez de usar conexões TCP externas.
13 InProcessServer InProcessConnProvider
14
15 //...
16}
1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4 InProcessConn() (net.Conn, error)
5}
- Voltando ao código: no client nats.go, o campo da struct de opções
InProcessServerfoi definido como a interfaceInProcessConnProvider, que realiza apenas a operaçãoInProcessConn.
1// nats-server/server/server.go
2
3// InProcessConn retorna uma conexão in-process com o servidor,
4// evitando a necessidade de um listener TCP para conectividade local
5// dentro do mesmo processo. Pode ser usada independentemente do
6// estado da opção DontListen.
7func (s *Server) InProcessConn() (net.Conn, error) {
8 pl, pr := net.Pipe()
9 if !s.startGoRoutine(func() {
10 s.createClientInProcess(pl)
11 s.grWG.Done()
12 }) {
13 pl.Close()
14 pr.Close()
15 return nil, fmt.Errorf("failed to create connection")
16 }
17 return pr, nil
18}
- No entanto, o tipo que é passado a ele é o
Serverdo nats-server, que executa diversas outras funcionalidades além de InProcessConn. - Isso ocorre porque, nesse contexto, o interesse do client se limita a saber se o objeto oferece a interface
InProcessConnou não; as demais funcionalidades não são relevantes para ele. - Por essa razão, o nats.go client utiliza apenas a interface
InProcessConnProvider, uma consumer defined interface que define somente a funcionalidadeInProcessConn() (net.Conn, error).
Conclusão
- Foram abordados de forma breve o embedded mode do NATS, seu funcionamento e a consumer defined interface do Go que pode ser observada no código do NATS.
- Espero que essas informações sejam úteis para quem utiliza o NATS com objetivos semelhantes, e encerro o texto por aqui.