GoSuda

Konfigurácia embedded NATS servera v aplikáciách Go a štruktúra komunikácie in-process

By prravda views ...

Začíname

O NATS

Softvérové aplikácie a služby potrebujú vymieňať dáta. NATS je infraštruktúra, ktorá umožňuje takúto výmenu dát, segmentovanú vo forme správ. Nazývame to „message oriented middleware“.

S NATS môžu vývojári aplikácií:

  • Bez námahy budovať distribuované a škálovateľné klient-serverové aplikácie.
  • Ukladať a distribuovať dáta v reálnom čase všeobecným spôsobom. Toto je možné flexibilne dosiahnuť naprieč rôznymi prostrediami, jazykmi, poskytovateľmi cloudu a lokálnymi systémami.

Čo je NATS, dokumentácia NATS

  • NATS je message broker napísaný v jazyku Go.

Embedded NATS

Ak je vaša aplikácia napísaná v Go a ak to zodpovedá vášmu prípadu použitia a scenárom nasadenia, môžete dokonca vložiť server NATS priamo do svojej aplikácie.

Embedding NATS, dokumentácia NATS

  • A NATS má aj jednu osobitosť: pre aplikácie napísané v Go podporuje embedded mode.
  • Inými slovami, namiesto bežného spôsobu, pri ktorom sa samostatný server brokera spustí a aplikácia s ním komunikuje prostredníctvom klienta, je možné samotný broker vložiť (embed) priamo do aplikácie napísanej v Go.

Výhody a prípady použitia vloženého NATS

  • Keďže existuje dobre vysvetľujúce video na Youtube, odkazujem na toto video namiesto podrobného výkladu.
  • Aj bez nasadenia samostatného servera message brokera je možné vytvoriť modulárnu monolitickú aplikáciu (modular monolith) a dosiahnuť separation of concerns, pričom sa zároveň využije možnosť vložiť NATS ako embedded. Navyše je umožnené nasadenie ako jediný binárny súbor (single binary deployment).
  • Okrem platforiem bez sieťového pripojenia (wasm) je to užitočné aj v aplikáciách typu offline-first.

Príklad z oficiálnej dokumentácie

 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    // Inicializácia nového servera s možnosťami
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Spustenie servera cez goroutine
22    go ns.Start()
23
24    // Počkajte, kým bude server pripravený na pripojenia
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Pripojenie k serveru
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Prihlásenie na odber témy
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Vypísanie dát správy
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Vypnutie servera (voliteľné)
45        ns.Shutdown()
46    })
47
48    // Publikovanie dát do témy
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Čakanie na vypnutie servera
52    ns.WaitForShutdown()
53}
  • Toto je príklad vloženého NATS, na ktorý odkazuje oficiálna dokumentácia NATS, no ak sa budeme riadiť týmto príkladovým kódom, komunikácia neprebieha v 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
  • Pomocou príkazu watch 'netstat -an | grep 127.0.0.1' sledujeme sieťovú komunikáciu smerujúcu na localhost (127.0.0.1) a pri spustení tohto go súboru príkazom go run . vidíme, ako pribúdajú nové sieťové požiadavky vychádzajúce z predvoleného portu NATS 4222.

Správne konfigurácie pre embedding mode

  • Aby komunikácia prebiehala v embedded mode tak, ako je zamýšľané, sú potrebné dve nasledujúce možnosti:

    • Klient: je potrebné nastaviť option InProcessServer.
    • Server: vo Server.Options je potrebné explicitne nastaviť flag DontListen na true.

  • Tieto časti nie sú oficiálne zdokumentované a začiatok tejto funkcionality je možné vystopovať v tomto PR.

    Tento PR pridáva tri veci:

    1. funkcia InProcessConn() pre Server, ktorá vytvára net.Pipe na získanie pripojenia k serveru NATS bez použitia TCP socketov
    2. option DontListen, ktorá serveru NATS hovorí, aby nepočúval na obvyklom TCP listeneri
    3. kanál startupComplete, ktorý sa zatvorí tesne pred spustením AcceptLoop, a readyForConnections naň bude čakať

    Hlavnou motiváciou je, že máme aplikáciu, ktorá môže bežať buď v monolitickom (jednoprocesovom) režime, alebo v polylitickom (viacprocesovom) režime. Chceli by sme pre zjednodušenie používať NATS v oboch režimoch, no monolitický režim musí podporovať rôzne platformy, kde otváranie socketových pripojení buď nedáva zmysel (mobilné zariadenia), alebo nie je vôbec možné (WASM). Tieto zmeny nám umožnia používať NATS úplne v rámci jedného procesu.

    Sprievodný PR nats-io/nats.go#774 pridáva podporu na strane klienta.

    Toto je môj prvý PR do tohto projektu, takže sa vopred ospravedlňujem, ak som niečo zjavné prehliadol.

    /cc @nats-io/core

Pracovný príklad pre 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		// Pre konfiguráciu vloženého servera NATS
14		// nastavte DontListen na true
15		DontListen: true,
16	}
17
18	// Inicializácia nového servera s možnosťami
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Spustenie servera cez goroutine
26	go ns.Start()
27
28	// Počkajte, kým bude server pripravený na pripojenia
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Pripojenie k serveru cez in-process pripojenie
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	// Prihlásenie na odber témy
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Vypísanie dát správy
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Vypnutie servera (voliteľné)
49		ns.Shutdown()
50	})
51
52	// Publikovanie dát do témy
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Čakanie na vypnutie servera
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
  • Teraz vidíme, že nedochádza k žiadnemu dodatočnému sieťovému skoku (network hop), ako sme zamýšľali.

Pod kapotou

TL;DR

diagram1

  • Ide o sekvenčný diagram znázorňujúci, aké funkcie sa interne vykonávajú a ako fungujú, keď sa tento kód spustí v main.go. Hlavné body sú nasledujúce:
    • Vďaka DontListen: true server vynecháva fázu počúvania klientov nazývanú AcceptLoop.
    • Ak je v možnostiach pripojenia klienta aktivovaná option InProcessServer, vytvorí sa in-memory pripojenie: pomocou net.Pipe sa vytvorí rúra a jej koniec sa klientovi vráti ako typ net.Conn.
    • Klient a server komunikujú v rámci tohto pripojenia v in-process režime.

Server

AcceptLoop

1// nats-server/server/server.go
2
3// Čakanie na klientov.
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}
  • Najprv, ak je DontListen true, server vynecháva fázu počúvania klientov nazývanú AcceptLoop.
 1// nats-server/server/server.go
 2
 3// AcceptLoop je exportovaná kvôli jednoduchšiemu testovaniu.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// Ak by sme skončili skôr, než sa listener správne nastaví, zabezpečte zatvorenie kanála.
 6	defer func() {
 7		if clr != nil {
 8			close(clr)
 9		}
10	}()
11
12	if s.isShuttingDown() {
13		return
14	}
15
16	// Snímka možností servera.
17	opts := s.getOpts()
18
19	// Nastavenie stavu, ktorý umožňuje vypnutie
20	s.mu.Lock()
21	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
22	l, e := natsListen("tcp", hp)
23	s.listenerErr = e
24	if e != nil {
25		s.mu.Unlock()
26		s.Fatalf("Error listening on port: %s, %q", hp, e)
27		return
28	}
29	s.Noticef("Listening for client connections on %s",
30		net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))
31
32	// Upozornenie, že TLS je povolené.
33	if opts.TLSConfig != nil {
34		s.Noticef("TLS required for client connections")
35		if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
36			s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
37		}
38	}
39
40	// Ak bol server spustený s RANDOM_PORT (-1), opts.Port by sa na začiatku tejto funkcie rovnal 0. Preto musíme získať skutočný port.
41	if opts.Port == 0 {
42		// Zapíšte získaný port späť do možností.
43		opts.Port = l.Addr().(*net.TCPAddr).Port
44	}
45
46	// Keď je port nastavený (ak bol nastavený na RANDOM), nastavte Host/Port v info servera buď z hodnôt Options, alebo z ClientAdvertise.
47	if err := s.setInfoHostPort(); err != nil {
48		s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
49		l.Close()
50		s.mu.Unlock()
51		return
52	}
53	// Uchovávajte URL adresy na pripojenie klientov. Môžeme ich neskôr potrebovať.
54	s.clientConnectURLs = s.getClientConnectURLs()
55	s.listener = l
56
57	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
58		func(_ error) bool {
59			if s.isLameDuckMode() {
60				// Signal that we are not accepting new clients
61				s.ldmCh <- true
62				// Now wait for the Shutdown...
63				<-s.quitCh
64				return true
65			}
66			return false
67		})
68	s.mu.Unlock()
69
70	// Oznámte volajúcemu, že sme pripravení
71	close(clr)
72	clr = nil
73}
  • Pre úplnosť, funkcia AcceptLoop vykonáva nasledujúce kroky. Ide o časti týkajúce sa sieťovej komunikácie, ako sú TLS alebo hostPort; pri in-process komunikácii nie sú potrebné, takže ich možno vynechať.

Client

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect sa pokúsi pripojiť k systému NATS.
 5// URL môže obsahovať používateľské meno a heslo, napr. nats://derek:pass@localhost:4222
 6// Podporované sú aj zoznamy oddelené čiarkami, napr. urlA, urlB.
 7// Možnosti sa najprv nastavia na predvolené hodnoty, ale je možné ich prepísať.
 8// Na pripojenie k websocket portu servera NATS použite schému `ws` alebo `wss`, napríklad
 9// `ws://localhost:8080`. Schémy websocket nemožno kombinovať s inými (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 slúži na vytvorenie prispôsobeného pripojenia.
 4type Options struct {
 5	// Url predstavuje jednu URL adresu servera NATS, ku ktorej sa klient pripojí. Ak je nastavená aj option Servers, stane sa prvým serverom v poli Servers.
 6	Url string
 7
 8	// InProcessServer predstavuje server NATS bežiaci v rámci rovnakého procesu. Ak je nastavený, pokúsime sa pripojiť priamo k serveru namiesto použitia externých TCP pripojení.
 9	InProcessServer InProcesConnProvider
10
11	//...
12}

1// nats-go/nats.go
2
3type InProcesConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Funkcia Connect, ktorá zabezpečuje pripojenie servera NATS a klienta NATS, prijíma URL klienta a connect options; štruktúra Options, ktorá tieto options zhromažďuje, obsahuje pole InProcessServer typu rozhrania InProcesConnProvider.
1// main.go of example code
2
3// Inicializácia nového servera s možnosťami
4ns, err := server.NewServer(opts)
5
6//...
7
8// Pripojenie k serveru cez in-process pripojenie
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
  • Keď klient NATS volá Connect a ako pole InProcessServer odovzdá nats.InProcessServer(ns),
1// nats-go/nats.go
2
3// InProcessServer je Option, ktorá sa pokúsi nadviazať priame spojenie so serverom NATS bežiacim v rámci procesu namiesto vytáčania cez TCP.
4func InProcessServer(server InProcesConnProvider) Option {
5	return func(o *Options) error {
6		o.InProcessServer = server
7		return nil
8	}
9}
  • hodnota option InProcessServer sa nahradí embedded serverom NATS,
 1// nats-go/nats.go
 2
 3// createConn sa pripojí k serveru a obalí zodpovedajúce štruktúry bufio. Pri existujúcom pripojení sa správne zachová.
 4func (nc *Conn) createConn() (err error) {
 5	if nc.Opts.Timeout < 0 {
 6		return ErrBadTimeout
 7	}
 8	if _, cur := nc.currentServer(); cur == nil {
 9		return ErrNoServers
10	}
11
12	// Ak máme referenciu na in-process server, nadviažeme pripojenie cez ňu.
13	if nc.Opts.InProcessServer != nil {
14		conn, err := nc.Opts.InProcessServer.InProcessConn()
15		if err != nil {
16			return fmt.Errorf("failed to get in-process connection: %w", err)
17		}
18		nc.conn = conn
19		nc.bindToNewConn()
20		return nil
21	}
22
23	//...
24}
  • toto rozhranie sa v funkcii createConn, ktorá vytvára pripojenie, použije tak, že ak option InProcessServer nie je nil (je platná), spustí InProcesConn z InProcessServer uvedeného v option
  • a zavolá implementáciu InProcessConn na strane servera.
 1// nats-server/server/server.go
 2
 3// InProcessConn vráti in-process pripojenie k serveru a eliminuje potrebu použitia TCP listeneru na lokálnu konektivitu v rámci toho istého procesu. Môže sa použiť bez ohľadu na stav option DontListen.
 4func (s *Server) InProcessConn() (net.Conn, error) {
 5	pl, pr := net.Pipe()
 6	if !s.startGoRoutine(func() {
 7		s.createClientInProcess(pl)
 8		s.grWG.Done()
 9	}) {
10		pl.Close()
11		pr.Close()
12		return nil, fmt.Errorf("failed to create connection")
13	}
14	return pr, nil
15}
  • Táto funkcia sa volá v go klientovi NATS, nats.go, keď InProcessServer nc (pripojenia NATS) nie je nil; vytvorí pripojenie (net.Conn) a zviaže ho s pripojením servera.

Consumer driven interface v jazyku Go

Typ implementuje rozhranie tým, že implementuje jeho metódy. Neexistuje žiadne explicitné deklarovanie úmyslu, žiadne kľúčové slovo „implements“. Implicitné rozhrania oddeľujú definíciu rozhrania od jeho implementácie, ktorá sa potom môže objaviť v ľubovoľnom balíku bez predchádzajúcej dohody.

Rozhrania sú implementované implicitne, A Tour of Go

Ak typ existuje len preto, aby implementoval rozhranie, a nikdy nebude mať exportované metódy nad rámec tohto rozhrania, nie je potrebné exportovať samotný typ.

Všeobecnosť, Effective Go

  • Tento návrh rozhraní dobre zachytáva to, čo sa v Go bežne nazýva consumer defined interface a structural typing (duck typing), preto by sme sa k tejto téme chceli vrátiť.
 1// nats-go/nats.go
 2
 3// Options slúži na vytvorenie prispôsobeného pripojenia.
 4type Options struct {
 5	// Url predstavuje jednu URL adresu servera NATS, ku ktorej sa klient pripojí. Ak je nastavená aj option Servers, stane sa prvým serverom v poli Servers.
 6	Url string
 7
 8	// InProcessServer predstavuje server NATS bežiaci v rámci rovnakého procesu. Ak je nastavený, pokúsime sa pripojiť priamo k serveru namiesto použitia externých TCP pripojení.
 9	InProcessServer InProcesConnProvider
10
11	//...
12}

1// nats-go/nats.go
2
3type InProcesConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Vráťme sa ku kódu. Klient nats.go definuje pole option InProcessServer ako rozhranie InProcesConnProvider, ktoré vykonáva iba InProcessConn.
 1// nats-server/server/server.go
 2
 3// InProcessConn vráti in-process pripojenie k serveru a eliminuje potrebu použitia TCP listeneru na lokálnu konektivitu v rámci toho istého procesu. Môže sa použiť bez ohľadu na stav option DontListen.
 4func (s *Server) InProcessConn() (net.Conn, error) {
 5	pl, pr := net.Pipe()
 6	if !s.startGoRoutine(func() {
 7		s.createClientInProcess(pl)
 8		s.grWG.Done()
 9	}) {
10		pl.Close()
11		pr.Close()
12		return nil, fmt.Errorf("failed to create connection")
13	}
14	return pr, nil
15}
  • Typom, ktorý sa do neho dosadí, je však Server z nats-server, ktorý okrem InProcessConn vykonáva aj mnoho ďalších funkcií.
  • Dôvod je ten, že v tejto situácii klienta zaujíma len to, či poskytuje rozhranie InProcessConn, a na ostatných veciach veľmi nezáleží.
  • Klient nats.go preto používa iba consumer defined rozhranie InProcessConnProvider, ktoré definuje len funkcionalitu InProcessConn() (net.Conn, error).

Záver

  • Stručne sme sa venovali embedded mode v NATS, jeho fungovaniu a consumer defined rozhraniam v Go, ktoré je možné vidieť v kóde NATS.
  • Dúfam, že tieto informácie pomôžu ľuďom, ktorí používajú NATS na podobné účely, a týmto článok končím.