GoSuda

Konfiguration af en indlejret NATS-server i Go-applikationer og arkitekturen for kommunikation i samme proces

By prravda views ...

Kom godt i gang

Om NATS

Softwareapplikationer og -tjenester skal udveksle data. NATS er en infrastruktur, der muliggør en sådan dataudveksling, opdelt i form af meddelelser. Vi kalder dette for en "message oriented middleware".

Med NATS kan applikationsudviklere:

  • Nemt bygge distribuerede og skalerbare klient-server-applikationer.
  • Lagre og distribuere data i realtid på en generel måde. Dette kan fleksibelt opnås på tværs af forskellige miljøer, sprog, cloududbydere og lokale systemer (on-premises).

What is NATS, NATS docs

  • NATS er en meddelelsesbroker bygget i Go.

Indlejret NATS

Hvis din applikation er skrevet i Go, og hvis det passer til dit use case og dine udrulningsscenarier, kan du endda indlejre en NATS-server i din applikation.

Embedding NATS, NATS docs

  • Der er også en særlig egenskab ved NATS: for applikationer skrevet i Go understøttes en embedded mode.
  • Det vil sige, at brokeren ikke behøver at køre som en separat brokerserver, som applikationens klienter kommunikerer med, som det er almindeligt for meddelelsesbrokere. I stedet kan selve brokeren indlejres (embed) i den applikation, der er skrevet i Go.

Fordele og use cases for indlejret NATS

  • Der findes en velforklaret Youtube-video, som vi henviser til i stedet for at gentage indholdet.
  • Uden at udrulle en separat meddelelsesbrokerserver kan man bygge en modular monolith-applikation og opnå separation of concerns, og samtidig drage fordel af at indlejre NATS. Derudover bliver single binary deployment mulig.
  • Det er nyttigt ikke kun i platforme uden netværk (wasm), men også i offline-first-applikationer.

Eksempel fra den officielle dokumentation

 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    // Opret en ny server med options
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Start serveren via goroutine
22    go ns.Start()
23
24    // Vent på, at serveren er klar til forbindelser
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Opret forbindelse til serveren
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Abonner på subjectet
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Udskriv meddelelsens data
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Luk serveren ned (valgfrit)
45        ns.Shutdown()
46    })
47
48    // Publicer data til subjectet
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Vent på, at serveren lukker ned
52    ns.WaitForShutdown()
53}
  • Dette er eksemplet på Embedded NATS, som den officielle NATS-dokumentation henviser til. Hvis man dog følger eksempelkoden som den er, sker kommunikationen ikke i 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
  • Ved at køre watch 'netstat -an | grep 127.0.0.1' kan man følge netværkstrafikken til localhost (127.0.0.1). Når man kører Go-filen med go run ., ser man nye netværksforespørgsler, der starter fra NATS' standardport 4222.

Korrekte konfigurationer for embedding mode

  • For at kommunikere i embedded mode som tilsigtet kræves to indstillinger:

    • Klient: InProcessServer-option skal angives.
    • Server: Flaget DontListen i Server.Options skal sættes til true.

  • Disse dele er ikke officielt dokumenteret, og begyndelsen på denne funktion kan findes i denne PR.

    This PR adds three things:

    1. InProcessConn() function to Server which builds a net.Pipe to get a connection to the NATS server without using TCP sockets
    2. DontListen option which tells the NATS server not to listen on the usual TCP listener
    3. startupComplete channel, which is closed right before we start AcceptLoop, and readyForConnections will 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

Fungerende eksempel for 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		// for configuring the embeded NATS server
14		// set DonListen as true
15		DontListen: true,
16	}
17
18	// Opret en ny server med options
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Start serveren via goroutine
26	go ns.Start()
27
28	// Vent på, at serveren er klar til forbindelser
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Opret forbindelse til serveren via in-process forbindelse
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	// Abonner på subjectet
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Udskriv meddelelsens data
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Luk serveren ned (valgfrit)
49		ns.Shutdown()
50	})
51
52	// Publicer data til subjectet
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Vent på, at serveren lukker ned
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 ses det, at der ikke opstår yderligere netværkshop, som tilsigtet.

Under overfladen

TL;DR

diagram1

  • Dette er et sequence diagram, der viser, hvilke funktioner der kører internt, og hvordan de arbejder sammen, når koden i main.go udføres. Hovedpunkterne er følgende:
    • Med DontListen: true springer serveren AcceptLoop-fasen over, som er client listening phase.
    • Hvis InProcessServer er aktiveret i klientens Connect-option, oprettes en in-memory-forbindelse. Der oprettes en pipe via net.Pipe, hvorefter den ene ende af pipen returneres til klienten som en net.Conn-type.
    • Klienten og serveren kommunikerer via denne forbindelse i in-process communication.

Server

AcceptLoop

1// nats-server/server/server.go
2
3// Vent på klienter.
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}
  • Hvis DontListen er true, springer serveren først AcceptLoop-fasen over, som er client listening phase.
 1// nats-server/server/server.go
 2
 3// AcceptLoop er eksporteret for lettere testning.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// Hvis vi skulle afslutte, før listeneren er sat korrekt op,
 6	// skal vi sørge for at lukke kanalen.
 7	defer func() {
 8		if clr != nil {
 9			close(clr)
10		}
11	}()
12
13	if s.isShuttingDown() {
14		return
15	}
16
17	// Tag et snapshot af serverens options.
18	opts := s.getOpts()
19
20	// Opsæt state, der muliggør nedlukning
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	// Underret om at TLS er aktiveret.
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	// Hvis serveren blev startet med RANDOM_PORT (-1), vil opts.Port være lig med
42	// 0 i begyndelsen af denne funktion. Så vi skal hente den faktiske port
43	if opts.Port == 0 {
44		// Skriv den fundne port tilbage til options.
45		opts.Port = l.Addr().(*net.TCPAddr).Port
46	}
47
48	// Nu hvor porten er sat (hvis den blev sat til RANDOM), sæt
49	// serverens info Host/Port med enten værdier fra 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	// Hold styr på klientens connect-URL'er. Vi har muligvis brug for dem senere.
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				// Signalér at vi ikke accepterer nye klienter
65				s.ldmCh <- true
66				// Vent nu på nedlukningen...
67				<-s.quitCh
68				return true
69			}
70			return false
71		})
72	s.mu.Unlock()
73
74	// Lad kalderen vide, at vi er klar
75	close(clr)
76	clr = nil
77}
  • Bemærk, at AcceptLoop-funktionen gennemfører følgende trin. TLS og hostPort hører til netværkskommunikation, og da de ikke er nødvendige ved in-process communication, kan de udelades.

Klient

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect forsøger at oprette forbindelse til NATS-systemet.
 5// URL'en kan indeholde username/password-oplysninger, f.eks. nats://derek:pass@localhost:4222.
 6// Kommaseparerede arrays understøttes også, f.eks. urlA, urlB.
 7// Options starter med standardværdierne, men kan overskrives.
 8// For at oprette forbindelse til en NATS-servers websocket-port, brug `ws`- eller `wss`-skemaet, f.eks.
 9// `ws://localhost:8080`. Bemærk, at websocket-skemaer ikke kan blandes med andre (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 bruges til at oprette en tilpasset forbindelse.
 4type Options struct {
 5	// Url angiver en enkelt NATS-server-URL, som klienten
 6	// forbinder til. Hvis Servers-optionen også er sat, bliver
 7	// den den første server i Servers-arrayet.
 8	Url string
 9
10	// InProcessServer angiver en NATS-server, der kører inden for
11	// samme proces. Hvis denne er sat, forsøger vi at oprette forbindelse
12	// direkte til serveren i stedet for at bruge eksterne TCP-forbindelser.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Connect-funktionen, der forbinder NATS-serveren med NATS-klienten, tager en client-URL og connect-options. Options-structen, der samler disse options, indeholder et felt ved navn InProcessServer af typen InProcessConnProvider interface.
1// main.go of example code
2
3// Opret en ny server med options
4ns, err := server.NewServer(opts)
5
6//...
7
8// Opret forbindelse til serveren via in-process forbindelse
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
  • Når NATS-klienten kalder Connect og sender nats.InProcessServer(ns) til feltet InProcessServer,
 1// nats-go/nats.go
 2
 3// InProcessServer er en Option, der forsøger at etablere en forbindelse til en NATS-server,
 4// som kører inden for processen, i stedet for at oprette forbindelse via TCP.
 5func InProcessServer(server InProcessConnProvider) Option {
 6	return func(o *Options) error {
 7		o.InProcessServer = server
 8		return nil
 9	}
10}
  • bliver option'ens InProcessServer erstattet med den indlejrede NATS-server,
 1// nats-go/nats.go
 2
 3// createConn opretter forbindelse til serveren og indpakker de relevante
 4// bufio-strukturer. Den håndterer korrekt tilfælde, hvor der allerede
 5// findes en forbindelse.
 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	// Hvis vi har en reference til en in-process server, etablerer vi en
15	// forbindelse via denne.
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}
  • Interfacet bruges i funktionen createConn, som opretter forbindelsen. Hvis InProcessServer-optionen ikke er nil (dvs. gyldig), kaldes InProcessConn på den InProcessServer, der er angivet i optionen,
 1// nats-server/server/server.go
 2
 3// InProcessConn returnerer en in-process forbindelse til serveren
 4// og undgår behovet for en TCP-listener til lokal forbindelse
 5// inden for samme proces. Dette kan bruges uanset
 6// tilstanden af DontListen-optionen.
 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 implementerede InProcessConn på serveren kaldes og udføres.
  • Denne funktion kaldes i nats.go, NATS' Go-klient, når InProcessServer på nc (NATS-forbindelsen) ikke er nil. Den opretter en forbindelse (net.Conn) og binder den til serverens forbindelse.

Forbrugerdefineret interface i Go

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.

Generality, Effective Go

  • Dette interfacedesign illustrerer godt det, man i Go ofte kalder consumer defined interface og structural typing (duck typing), og derfor vil jeg også introducere dette emne.
 1// nats-go/nats.go
 2
 3// Options kan bruges til at oprette en tilpasset forbindelse.
 4type Options struct {
 5	// Url angiver en enkelt NATS-server-URL, som klienten
 6	// forbinder til. Hvis Servers-optionen også er sat, bliver
 7	// den den første server i Servers-arrayet.
 8	Url string
 9
10	// InProcessServer angiver en NATS-server, der kører inden for
11	// samme proces. Hvis denne er sat, forsøger vi at oprette forbindelse
12	// direkte til serveren i stedet for at bruge eksterne TCP-forbindelser.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Lad os vende tilbage til koden. I nats.go-klienten er feltet InProcessServer i option-structen defineret som interfacet InProcessConnProvider, som kun udfører InProcessConn.
 1// nats-server/server/server.go
 2
 3// InProcessConn returnerer en in-process forbindelse til serveren
 4// og undgår behovet for en TCP-listener til lokal forbindelse
 5// inden for samme proces. Dette kan bruges uanset
 6// tilstanden af DontListen-optionen.
 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 type, der sendes ind, er dog Server fra nats-server, som udfører mange flere funktioner end blot InProcessConn.
  • Det skyldes, at klientens interesse i denne situation kun er, om serveren leverer interfacet InProcessConn eller ej; de øvrige funktioner er ikke særlig relevante for klienten.
  • Derfor bruger nats.go-klienten kun et consumer defined interface ved navn InProcessConnProvider, som kun definerer funktionen InProcessConn() (net.Conn, error).

Konklusion

  • Vi har kort gennemgået NATS' embedded mode og dets funktionsmåde samt det consumer defined interface i Go, som kan ses i NATS' kode.
  • Jeg håber, at denne information er nyttig for dem, der bruger NATS til lignende formål, og hermed afslutter jeg denne artikel.