GoSudaConfigurazione di un server NATS integrato in un'applicazione Go e struttura della comunicazione in-process
By prravda views ...
Per iniziare
Informazioni su NATS
Le applicazioni software e i servizi devono scambiarsi dati. NATS è un'infrastruttura che consente tale scambio di dati, suddivisi sotto forma di messaggi. La definiamo "message oriented middleware".
Con NATS, gli sviluppatori di applicazioni possono:
- Costruire in modo semplice applicazioni client-server distribuite e scalabili.
- Memorizzare e distribuire dati in tempo reale in modo generale. Ciò può essere realizzato in modo flessibile across vari ambienti, linguaggi, provider cloud e sistemi on-premises.
- NATS è un message broker costruito in Go.
NATS incorporato
Se la tua applicazione è scritta in Go e se si adatta al tuo caso d'uso e agli scenari di deployment, puoi persino incorporare un server NATS all'interno della tua applicazione.
- Una particolarità di NATS è che, per le applicazioni costruite in Go, supporta la modalità embedded.
- In altre parole, anziché il metodo tipico dei message broker, ovvero avviare un server broker separato e comunicare con esso tramite i client dell'applicazione, è possibile incorporare il broker stesso (embed) nell'applicazione realizzata in Go.
Vantaggi e casi d'uso di NATS incorporato
- Poiché esiste un video su YouTube ben curato, si rimanda al link del video.
- Anche senza distribuire un server di message broker separato, è possibile creare un'applicazione modular monolith e ottenere la separation of concerns, traendo vantaggio dalla possibilità di incorporare NATS in modalità embedded. Inoltre, diventa possibile il single binary deployment.
- Può essere utile non solo per la platform with no network (wasm), ma anche per le offline-first application.
Esempio nella documentazione ufficiale
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
14 // Inizializza un nuovo server con le opzioni
15 ns, err := server.NewServer(opts)
16
17 if err != nil {
18 panic(err)
19 }
20
21 // Avvia il server tramite goroutine
22 go ns.Start()
23
24 // Attende che il server sia pronto per le connessioni
25 if !ns.ReadyForConnections(4 * time.Second) {
26 panic("not ready for connection")
27 }
28
29 // Connessione al server
30 nc, err := nats.Connect(ns.ClientURL())
31
32 if err != nil {
33 panic(err)
34 }
35
36 subject := "my-subject"
37
38 // Sottoscrizione al subject
39 nc.Subscribe(subject, func(msg *nats.Msg) {
40 // Stampa i dati del messaggio
41 data := string(msg.Data)
42 fmt.Println(data)
43
44 // Arresta il server (opzionale)
45 ns.Shutdown()
46 })
47
48 // Pubblica dati sul subject
49 nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51 // Attende l'arresto del server
52 ns.WaitForShutdown()
53}
- Si tratta dell'esempio di NATS incorporato riportato dalla documentazione ufficiale di NATS, ma seguendo il codice di esempio la comunicazione non avviene in modalità embedding.
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
- Eseguendo
watch 'netstat -an | grep 127.0.0.1'per monitorare la rete che transita su localhost (127.0.0.1) e avviando il file go congo run ., si può osservare l'aggiunta di nuove richieste di rete che partono dalla porta predefinita di NATS,4222.
Configurazioni corrette per la modalità embedding
Per comunicare nella modalità embedded come previsto sono necessarie le seguenti due opzioni.
- Client: è necessario inserire l'opzione
InProcessServer. - Server: è necessario impostare esplicitamente il flag
DontListendiServer.Optionsatrue.
- Client: è necessario inserire l'opzione
Tali aspetti non sono ufficialmente documentati; l'origine di questa funzionalità è rintracciabile nella seguente PR.
This PR adds three things:
InProcessConn()function toServerwhich builds anet.Pipeto get a connection to the NATS server without using TCP socketsDontListenoption which tells the NATS server not to listen on the usual TCP listenerstartupCompletechannel, which is closed right before we startAcceptLoop, andreadyForConnectionswill wait for it
The main motivation for this is that we have an application that can run either in a monolith (single-process) mode or a polylith (multi-process) mode. We'd like to be able to use NATS for both modes for simplicity, but the monolith mode has to be able to cater for a variety of platforms where opening socket connections either doesn't make sense (mobile) or just isn't possible (WASM). These changes will allow us to use NATS entirely in-process instead.
An accompanying PR nats-io/nats.go#774 adds support to the client side.
This is my first PR to this project so apologies in advance if I've missed anything obvious anywhere.
/cc @nats-io/core
Esempio funzionante per la modalità embedded
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 // per configurare il server NATS incorporato
14 // impostare DonListen a true
15 DontListen: true,
16 }
17
18 // Inizializza un nuovo server con le opzioni
19 ns, err := server.NewServer(opts)
20
21 if err != nil {
22 panic(err)
23 }
24
25 // Avvia il server tramite goroutine
26 go ns.Start()
27
28 // Attende che il server sia pronto per le connessioni
29 if !ns.ReadyForConnections(10 * time.Second) {
30 panic("not ready for connection")
31 }
32
33 // Connessione al server tramite connessione 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 // Sottoscrizione al subject
43 nc.Subscribe(subject, func(msg *nats.Msg) {
44 // Stampa i dati del messaggio
45 data := string(msg.Data)
46 fmt.Println(data)
47
48 // Arresta il server (opzionale)
49 ns.Shutdown()
50 })
51
52 // Pubblica dati sul subject
53 nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55 // Attende l'arresto del server
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
- Ora si può osservare che non si verifica alcun network hop aggiuntivo, come previsto.
Sotto il cofano
TL;DR
- Si tratta di un sequence diagram che mostra quali funzioni vengono eseguite e in che modo quando il codice viene eseguito da
main.go; i punti principali sono i seguenti.- Tramite
DontListen: true, il server omette la fase di ascolto dei client denominataAcceptLoop. - Se l'opzione
InProcessServerdel Connect del client è attiva, viene creata una connessione in-memory tramitenet.Pipe, e l'estremità della pipe viene restituita al client come tiponet.Conn. - Client e server comunicano in-process attraverso tale connessione.
- Tramite
Server
AcceptLoop
1// nats-server/server/server.go
2
3// Attende i client.
4if !opts.DontListen {
5 s.AcceptLoop(clientListenReady)
6}
- Innanzitutto, quando
DontListenè true, la fase di ascolto dei client denominataAcceptLoopviene omessa.
1// nats-server/server/server.go
2
3// AcceptLoop è esportata per facilitare i test.
4func (s *Server) AcceptLoop(clr chan struct{}) {
5 // Se uscissimo prima che il listener sia configurato correttamente,
6 // assicuriamoci di chiudere il canale.
7 defer func() {
8 if clr != nil {
9 close(clr)
10 }
11 }()
12
13 if s.isShuttingDown() {
14 return
15 }
16
17 // Cattura uno snapshot delle opzioni del server.
18 opts := s.getOpts()
19
20 // Configura lo stato che consente lo 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 // Segnala che TLS è abilitato.
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 il server è stato avviato con RANDOM_PORT (-1), opts.Port sarebbe uguale
42 // a 0 all'inizio di questa funzione. Quindi dobbiamo ottenere la porta effettiva
43 if opts.Port == 0 {
44 // Riscrive la porta risolta nelle opzioni.
45 opts.Port = l.Addr().(*net.TCPAddr).Port
46 }
47
48 // Ora che la porta è stata impostata (se era RANDOM), imposta
49 // Host/Port delle INFO del server con i valori di Options o
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 // Tiene traccia degli URL di connessione dei client. Potrebbero servire più avanti.
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 // Segnala che non stiamo accettando nuovi client
65 s.ldmCh <- true
66 // Ora attende lo Shutdown...
67 <-s.quitCh
68 return true
69 }
70 return false
71 })
72 s.mu.Unlock()
73
74 // Comunica al chiamante che siamo pronti
75 close(clr)
76 clr = nil
77}
- Va notato che la funzione AcceptLoop esegue i passaggi seguenti. Le parti relative alla comunicazione di rete, come
TLSohostPort, non sono necessarie per la comunicazione in-process e possono quindi essere omesse.
Client
InProcessServer
1
2// nats-go/nats.go
3
4// Connect tenta di connettersi al sistema NATS.
5// L'url può contenere la semantica username/password. es. nats://derek:pass@localhost:4222
6// Sono supportati anche array separati da virgole, es. urlA, urlB.
7// Le opzioni partono dai valori predefiniti ma possono essere sovrascritte.
8// Per connettersi alla porta websocket di un server NATS, usare lo schema `ws` o `wss`, come
9// `ws://localhost:8080`. Si noti che gli schemi websocket non possono essere combinati con altri (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 può essere usato per creare una connessione personalizzata.
4type Options struct {
5 // Url rappresenta un singolo url del server NATS a cui il client
6 // si connetterà. Se è impostata anche l'opzione Servers, diventa
7 // il primo server nell'array Servers.
8 Url string
9
10 // InProcessServer rappresenta un server NATS in esecuzione all'interno
11 // dello stesso processo. Se è impostato, tenteremo di connetterci
12 // direttamente al server anziché usare connessioni TCP esterne.
13 InProcessServer InProcessConnProvider
14
15 //...
16}
1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4 InProcessConn() (net.Conn, error)
5}
- La funzione
Connect, che realizza la connessione tra il server NATS e il client NATS, accetta l'URL del client e le opzioni di connessione; la struct Options, che raccoglie tali opzioni, contiene il campoInProcessServerdi tipo interfaceInProcessConnProvider.
1// main.go of example code
2
3// Inizializza un nuovo server con le opzioni
4ns, err := server.NewServer(opts)
5
6//...
7
8// Connessione al server tramite connessione in-process
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
- Quando il client NATS esegue Connect e passa
nats.InProcessServer(ns)nel campoInProcessServer,
1// nats-go/nats.go
2
3// InProcessServer è un'Option che tenta di stabilire una connessione verso un server NATS
4// in esecuzione all'interno del processo, anziché effettuare il dial tramite TCP.
5func InProcessServer(server InProcessConnProvider) Option {
6 return func(o *Options) error {
7 o.InProcessServer = server
8 return nil
9 }
10}
- l'InProcessServer dell'opzione viene sostituito con il server NATS incorporato,
1// nats-go/nats.go
2
3// createConn si connette al server e avvolge le
4// strutture bufio appropriate. Gestisce correttamente il caso in cui
5// sia già presente una connessione esistente.
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 abbiamo un riferimento a un server in-process, stabiliamo una
15 // connessione usando quello.
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}
- Questa interface viene utilizzata nella funzione
createConn, che crea la connessione: se l'opzioneInProcessServernon è nil (quindi valida), viene eseguito il metodoInProcessConndell'InProcessServer contenuto nell'opzione,
1// nats-server/server/server.go
2
3// InProcessConn restituisce una connessione in-process al server,
4// evitando la necessità di usare un listener TCP per la connettività locale
5// all'interno dello stesso processo. Può essere usato indipendentemente
6// dallo stato dell'opzione 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}
- Viene quindi invocato il
InProcessConnimplementato nel server. - Questa funzione viene chiamata da
nats.go, il client Go di NATS, quando l'InProcessServerdell'nc (connessione NATS) non è nil: crea la connessione (net.Conn) e la associa alla connessione del server.
Interface Go definita dal consumer
A type implements an interface by implementing its methods. There is no explicit declaration of intent, no "implements" keyword. Implicit interfaces decouple the definition of an interface from its implementation, which could then appear in any package without prearrangement.
Interfaces are implemented implicitly, A Tour of Go
If a type exists only to implement an interface and will never have exported methods beyond that interface, there is no need to export the type itself.
- Questo design delle interface rispecchia bene le cosiddette consumer defined interface e lo structural typing (duck typing) tipici di Go, per cui si è deciso di presentare anche questo argomento.
1// nats-go/nats.go
2
3// Options può essere usato per creare una connessione personalizzata.
4type Options struct {
5 // Url rappresenta un singolo url del server NATS a cui il client
6 // si connetterà. Se è impostata anche l'opzione Servers, diventa
7 // il primo server nell'array Servers.
8 Url string
9
10 // InProcessServer rappresenta un server NATS in esecuzione all'interno
11 // dello stesso processo. Se è impostato, tenteremo di connetterci
12 // direttamente al server anziché usare connessioni TCP esterne.
13 InProcessServer InProcessConnProvider
14
15 //...
16}
1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4 InProcessConn() (net.Conn, error)
5}
- Torniamo al codice. Nel client nats.go, il campo dell'opzione
InProcessServerè definito come interfaceInProcessConnProvider, che prevede soloInProcessConn.
1// nats-server/server/server.go
2
3// InProcessConn restituisce una connessione in-process al server,
4// evitando la necessità di usare un listener TCP per la connettività locale
5// all'interno dello stesso processo. Può essere usato indipendentemente
6// dallo stato dell'opzione 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}
- Tuttavia, il tipo che viene inserito è il
Serverdi nats-server, che svolge molte altre funzioni oltre aInProcessConn. - Questo perché, in questo contesto, l'interesse del client riguarda solo se è stata fornita l'interface
InProcessConno meno, mentre il resto non è rilevante. - Pertanto, il client nats.go utilizza solo l'interface
InProcessConnProvider, un'interface definita dal consumer che prevede unicamente la funzionalitàInProcessConn() (net.Conn, error).
Conclusione
- Abbiamo trattato in sintesi la modalità embedded di NATS, il suo funzionamento e la consumer defined interface di Go che si può osservare nel codice di NATS.
- Auspico che queste informazioni siano utili a chi utilizza NATS con questi scopi, e con questo concludo l'articolo.