GoSuda

Konfiguration av inbäddad NATS-server i Go-applikationer och arkitekturen för kommunikation inom processen

By prravda views ...

Komma igång

Om NATS

Programvaruapplikationer och tjänster behöver utbyta data. NATS är en infrastruktur som möjliggör sådant datautbyte, uppdelat i form av meddelanden. Vi kallar detta för "message oriented middleware".

Med NATS kan applikationsutvecklare:

  • Utan ansträngning bygga distribuerade och skalbara klient-server-applikationer.
  • Lagra och distribuera data i realtid på ett generellt sätt. Detta kan på ett flexibelt sätt uppnås över olika miljöer, språk, molnleverantörer och lokala system (on-premises).

What is NATS, NATS docs

  • NATS är en meddelandeförmedlare (message broker) implementerad i Go.

Inbäddad NATS

Om din applikation är skriven i Go, och om det passar ditt användningsfall och dina driftscenarier, kan du till och med bädda in en NATS-server inuti din applikation.

Embedding NATS, NATS docs

  • Det finns även en särskild egenskap hos NATS: för applikationer skrivna i Go stöds ett embedded-läge.
  • Det innebär att meddelandeförmedlaren, i stället för det vanliga sättet där en separat förmedlarserver startas och applikationens klienter kommunicerar med den servern, kan inbäddas (embed) direkt i den applikation som är skriven i Go.

Fördelar och användningsfall för inbäddad NATS

  • Det finns en väl genomarbetad Youtube-video, och vi hänvisar därför till videons länk.
  • Även utan att distribuera en separat meddelandeförmedlarserver kan man bygga en modulär monolitisk applikation (modular monolith) och uppnå separation av ansvarsområden (separation of concerns) samt dra nytta av fördelen att NATS kan inbäddas. Dessutom blir distribution som en enda binärfil (single binary deployment) möjlig.
  • Det är användbart inte bara för plattformar utan nätverk (wasm) utan även för offline-first-applikationer.

Exempel från den officiella dokumentationen

 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    // Initialisera en ny server med options
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Starta servern via goroutine
22    go ns.Start()
23
24    // Vänta tills servern är redo för anslutningar
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Anslut till servern
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Prenumerera på subjectet
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Skriv ut meddelandedata
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Stäng av servern (valfritt)
45        ns.Shutdown()
46    })
47
48    // Publicera data till subjectet
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Vänta på att servern stängs av
52    ns.WaitForShutdown()
53}
  • Detta är exemplet på Embedded NATS som NATS officiella dokumentation länkar till, men om man följer exempelkoden så sker kommunikationen inte i embedding-läge.
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
  • Om man använder kommandot watch 'netstat -an | grep 127.0.0.1' för att följa nätverkstrafiken till localhost (127.0.0.1) och samtidigt kör den Go-fil i fråga med go run . ser man att nya nätverksanrop tillkommer, med start från NATS standardport 4222.

Korrekta konfigurationer för embedding-läge

  • För att kommunicera i embedded-läge som avsett krävs följande två alternativ.

    • Klient: InProcessServer-optionen måste anges.
    • Server: flaggan DontListen i Server.Options måste sättas explicit till true.

  • Dessa delar är inte officiellt dokumenterade, men ursprunget till funktionen kan spåras via denna PR.

    Denna PR lägger till tre saker:

    1. Funktionen InProcessConn() till Server, som bygger en net.Pipe för att få en anslutning till NATS-servern utan att använda TCP-sockets
    2. Optionen DontListen, som talar om för NATS-servern att inte lyssna på den vanliga TCP-listenern
    3. Kanalen startupComplete, som stängs precis innan vi startar AcceptLoop, och readyForConnections väntar på den

    Den främsta motivationen är att vi har en applikation som kan köras antingen i monolitiskt läge (en process) eller polylitiskt läge (flera processer). Vi vill av enkelhetsskäl kunna använda NATS i båda lägena, men monolitläget måste kunna fungera på en mängd plattformar där öppning av socketanslutningar antingen inte är meningsfullt (mobil) eller helt omöjligt (WASM). Dessa ändringar gör det möjligt att använda NATS helt inom processen i stället.

    En medföljande PR nats-io/nats.go#774 lägger till stöd på klientsidan.

    Detta är mitt första PR till projektet, så jag ber om ursäkt i förväg om jag har missat något uppenbart.

    /cc @nats-io/core

Fungerande exempel för embedded-läge

 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		// för att konfigurera den inbäddade NATS-servern
14		// sätt DonListen till true
15		DontListen: true,
16	}
17
18	// Initialisera en ny server med options
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Starta servern via goroutine
26	go ns.Start()
27
28	// Vänta tills servern är redo för anslutningar
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Anslut till servern via in-process-anslutning
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	// Prenumerera på subjectet
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Skriv ut meddelandedata
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Stäng av servern (valfritt)
49		ns.Shutdown()
50	})
51
52	// Publicera data till subjectet
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Vänta på att servern stängs av
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
  • Nu kan man se att inga extra nätverkshopp uppstår, precis som avsett.

Under huven

TL;DR

diagram1

  • Detta är ett sekvensdiagram som visar vilka funktioner som anropas och hur de fungerar internt när koden körs från main.go. Huvudpunkterna sammanfattas nedan.
    • Med DontListen: true hoppar servern över client-lyssningsfasen AcceptLoop.
    • Om InProcessServer är aktiverad bland klientens Connect-options skapas en in-memory-anslutning: en pipe skapas med net.Pipe, och pipens ena ände returneras till klienten som en net.Conn.
    • Klienten och servern kommunicerar sedan in-process via denna anslutning.

Server

AcceptLoop

1// nats-server/server/server.go
2
3// Vänta på klienter.
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}
  • Om DontListen är true hoppar servern först över client-lyssningsfasen AcceptLoop.
 1// nats-server/server/server.go
 2
 3// AcceptLoop exporteras för enklare testning.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// Om vi skulle avsluta innan lyssnaren är korrekt konfigurerad,
 6	// se till att kanalen stängs.
 7	defer func() {
 8		if clr != nil {
 9			close(clr)
10		}
11	}()
12
13	if s.isShuttingDown() {
14		return
15	}
16
17	// Ta en ögonblicksbild av serverns options.
18	opts := s.getOpts()
19
20	// Konfigurera tillstånd som gör det möjligt att stänga av
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	// Meddela att TLS är aktiverat.
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	// Om servern startades med RANDOM_PORT (-1) är opts.Port lika med
42	// 0 i början av denna funktion. Därför måste vi hämta den faktiska porten
43	if opts.Port == 0 {
44		// Skriv tillbaka den upplösta porten till options.
45		opts.Port = l.Addr().(*net.TCPAddr).Port
46	}
47
48	// Nu när porten har satts (om den var RANDOM), sätt
49	// serverns info Host/Port med antingen värden från Options eller
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	// Håll reda på klientanslutnings-URL:erna. Vi kan behöva dem senare.
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				// Signalera att vi inte längre accepterar nya klienter
65				s.ldmCh <- true
66				// Vänta nu på Shutdown...
67				<-s.quitCh
68				return true
69			}
70			return false
71		})
72	s.mu.Unlock()
73
74	// Låt anroparen veta att vi är redo
75	close(clr)
76	clr = nil
77}
  • Funktionen AcceptLoop genomför följande steg. Delarna som rör nätverkskommunikation, såsom TLS och hostPort, är onödiga vid in-process-kommunikation och kan därför utelämnas.

Klient

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect försöker ansluta till NATS-systemet.
 5// URL:en kan innehålla användarnamn/lösenord, t.ex. nats://derek:pass@localhost:4222.
 6// Kommaseparerade arrayer stöds också, t.ex. urlA, urlB.
 7// Options börjar med standardvärden men kan åsidosättas.
 8// För att ansluta till en NATS-servers websocket-port, använd schemat `ws` eller `wss`, t.ex.
 9// `ws://localhost:8080`. Observera att websocket-scheman inte kan blandas med andra (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 kan användas för att skapa en anpassad anslutning.
 4type Options struct {
 5	// Url representerar en enskild NATS-server-URL till vilken klienten
 6	// ansluter. Om Servers-optionen också är satt blir den
 7	// första servern i Servers-arrayen.
 8	Url string
 9
10	// InProcessServer representerar en NATS-server som körs inom
11	// samma process. Om denna är satt försöker vi ansluta
12	// direkt till servern i stället för att använda externa TCP-anslutningar.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Funktionen Connect, som upprättar anslutningen mellan NATS-servern och NATS-klienten, tar emot client-URL och connect-options. Options-structen som samlar dessa options innehåller fältet InProcessServer av interface-typen InProcessConnProvider.
1// main.go i exempelkoden
2
3// Initialisera en ny server med options
4ns, err := server.NewServer(opts)
5
6//...
7
8// Anslut till servern via in-process-anslutning
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
  • När NATS-klienten anropar Connect och skickar nats.InProcessServer(ns) till fältet InProcessServer
 1// nats-go/nats.go
 2
 3// InProcessServer är en Option som försöker etablera en riktning till en NATS-server
 4// som körs inom processen i stället för att ringa via TCP.
 5func InProcessServer(server InProcessConnProvider) Option {
 6	return func(o *Options) error {
 7		o.InProcessServer = server
 8		return nil
 9	}
10}
  • ersätts option-fältet InProcessServer med den inbäddade NATS-servern,
 1// nats-go/nats.go
 2
 3// createConn ansluter till servern och omsluter lämpliga
 4// bufio-strukturer. Den hanterar rätt när en befintlig
 5// anslutning redan finns.
 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	// Om vi har en referens till en in-process-server, etablera en
15	// anslutning med den.
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}
  • och interfacet anropas i funktionen createConn, som skapar anslutningen. Om optionen InProcessServer är skild från nil (giltig) körs InProcessConn för InProcessServer i option.
 1// nats-server/server/server.go
 2
 3// InProcessConn returnerar en in-process-anslutning till servern,
 4// vilket undviker behovet av en TCP-listener för lokal anslutning
 5// inom samma process. Detta kan användas oavsett
 6// tillståndet för optionen 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}
  • Den implementerade InProcessConn i servern anropas och körs.
  • Denna funktion anropas i nats.go, NATS Go-klienten, när InProcessServer för nc (NATS-anslutningen) är skild från nil. Den skapar en anslutning (net.Conn) och binder den till serverns anslutning.

Konsumentdriven interface i Go

En typ implementerar ett interface genom att implementera dess metoder. Det finns ingen explicit deklaration av avsikt, inget nyckelord "implements". Implicita interfaces frikopplar definitionen av ett interface från dess implementation, som då kan förekomma i vilket paket som helst utan förhandsarrangemang.

Interfaces are implemented implicitly, A Tour of Go

Om en typ endast existerar för att implementera ett interface och aldrig kommer att ha exporterade metoder utöver detta interface, finns det inget behov av att exportera själva typen.

Generality, Effective Go

  • Denna interface-design förkroppsligar väl det som i Go ofta kallas consumer-defined interface och structural typing (duck typing), och därför vill vi även introducera detta ämne.
 1// nats-go/nats.go
 2
 3// Options kan användas för att skapa en anpassad anslutning.
 4type Options struct {
 5	// Url representerar en enskild NATS-server-URL till vilken klienten
 6	// ansluter. Om Servers-optionen också är satt blir den
 7	// första servern i Servers-arrayen.
 8	Url string
 9
10	// InProcessServer representerar en NATS-server som körs inom
11	// samma process. Om denna är satt försöker vi ansluta
12	// direkt till servern i stället för att använda externa TCP-anslutningar.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Låt oss återgå till koden. I NATS Go-klienten är fältet InProcessServer i option-structen definierat som interfacet InProcessConnProvider, som endast innehåller InProcessConn.
 1// nats-server/server/server.go
 2
 3// InProcessConn returnerar en in-process-anslutning till servern,
 4// vilket undviker behovet av en TCP-listener för lokal anslutning
 5// inom samma process. Detta kan användas oavsett
 6// tillståndet för optionen 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}
  • Typen som skickas in är dock NATS-serverns Server, som utför många fler funktioner än bara InProcessConn.
  • Det beror på att klientens intresse i detta sammanhang endast gäller om den tillhandahåller interfacet InProcessConn, medan allt annat inte är av större betydelse.
  • Därför använder NATS Go-klienten endast interfacet InProcessConnProvider, ett consumer-defined interface som definierar enbart funktionen InProcessConn() (net.Conn, error).

Slutsats

  • Vi har kort behandlat NATS embedded-läge och hur det fungerar, samt det consumer-defined interface i Go som kan observeras i NATS kod.
  • Vi hoppas att denna information är till hjälp för dem som använder NATS i ett liknande syfte, och med detta avslutar vi artikeln.