Komme i gang

Om NATS

Programvareapplikasjoner og -tjenester må utveksle data. NATS er en infrastruktur som muliggjør slik datautveksling, segmentert i form av meldinger. Vi kaller dette for «message oriented middleware».

Med NATS kan applikasjonsutviklere:

• Uanstrengt bygge distribuerte og skalerbare klient-server-applikasjoner.
• Lagre og distribuere data i sanntid på en generell måte. Dette kan oppnås fleksibelt på tvers av ulike miljøer, programmeringsspråk, skyleverandører og lokale systemer.

Hva er NATS, NATS-dokumentasjon

• NATS er en meldingsmegler (message broker) implementert i Go.

Innebygd NATS

Dersom applikasjonen din er skrevet i Go, og dersom dette passer til bruksområdet og distribusjonsscenarioene dine, kan du til og med bygge inn en NATS-server i applikasjonen.

Innebygging av NATS, NATS-dokumentasjon

• I tillegg har NATS en særegenhet: for applikasjoner skrevet i Go støttes en embedded mode (innebygd modus).
• Med andre ord betyr dette at meldingsmegleren i seg selv kan innebygges (embed) i en applikasjon skrevet i Go, i stedet for den vanlige ordningen der en separat meldingsmeglerserver startes og kommuniseres med via applikasjonens klienter.

Fordeler og bruksområder for innebygd NATS

• Det finnes en godt forklart YouTube-video, og vi viser derfor til videoen i stedet for å gjengi innholdet her.
• Selv uten å distribuere en egen meldingsmeglerserver kan man bygge en modulær monolitt (modular monolith) applikasjon og oppnå separation of concerns, og dermed dra nytte av fordelen ved å bygge inn NATS. I tillegg muliggjøres distribusjon som én enkelt binærfil (single binary deployment).
• Den er nyttig ikke bare på plattformer uten nettverk (wasm), men også i offline-first-applikasjoner.

Eksempel fra den offisielle dokumentasjonen

 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    // Initialiser ny server med alternativer
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Start serveren via en goroutine
22    go ns.Start()
23
24    // Vent til serveren er klar for tilkoblinger
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Koble 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å emnet
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Skriv ut meldingsdataene
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Stopp serveren (valgfritt)
45        ns.Shutdown()
46    })
47
48    // Publiser data til emnet
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Vent på at serveren stopper
52    ns.WaitForShutdown()
53}

• Dette er et eksempel på Embedded NATS som er lenket fra den offisielle NATS-dokumentasjonen, men dersom man følger eksempelkoden slavisk, skjer kommunikasjonen ikke i embedding-modus.

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 å bruke kommandoen watch 'netstat -an | grep 127.0.0.1' for å overvåke nettverkstrafikken til localhost (127.0.0.1), og samtidig kjøre filen med go run ., ser man nye nettverksforespørsler som går ut fra 4222, NATS' standardport.

Korrekte konfigurasjoner for embedding-modus

• For å kommunisere i embedded mode slik det er tenkt, kreves to alternativer:

  • Klient: alternativet InProcessServer må settes.
  • Server: flagget DontListen må settes til true i Server.Options.

• Disse detaljene er ikke formelt dokumentert, men opprinnelsen til funksjonen kan spores gjennom denne PR-en.

  Denne PR-en legger til tre ting:

  1. Funksjonen InProcessConn() for Server, som bygger en net.Pipe for å få en forbindelse til NATS-serveren uten bruk av TCP-sockets
  2. Alternativet DontListen, som forteller NATS-serveren at den ikke skal lytte på den vanlige TCP-lytteren
  3. Kanalen startupComplete, som lukkes rett før vi starter AcceptLoop, og readyForConnections vil vente på den

  Hovedmotivasjonen for dette er at vi har en applikasjon som kan kjøres enten i monolittmodus (én prosess) eller polylittmodus (flere prosesser). Vi ønsker å kunne bruke NATS i begge modi for enkelhets skyld, men monolittmodus må kunne støtte en rekke plattformer der åpning av socket-forbindelser enten ikke gir mening (mobil) eller rett og slett ikke er mulig (WASM). Disse endringene gjør det mulig å bruke NATS helt og holdent innenfor prosessen i stedet.

  En tilhørende PR nats-io/nats.go#774 legger til støtte på klientsiden.

  Dette er min første PR til dette prosjektet, så jeg beklager på forhånd dersom jeg har oversett noe åpenbart et sted.

  /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 å konfigurere den innebygde NATS-serveren
14		// sett DonListen til true
15		DontListen: true,
16	}
17
18	// Initialiser ny server med alternativer
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Start serveren via en goroutine
26	go ns.Start()
27
28	// Vent til serveren er klar for tilkoblinger
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Koble til serveren via en 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å emnet
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Skriv ut meldingsdataene
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Stopp serveren (valgfritt)
49		ns.Shutdown()
50	})
51
52	// Publiser data til emnet
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Vent på at serveren stopper
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...ingen ekstra logger
4

• Nå ser man at det ikke oppstår noe ekstra nettverkshopp, slik det var ment.

Under panseret

TL;DR

• Dette er et sekvensdiagram som viser hvilke funksjoner som kjøres internt, og hvordan de virker sammen, når koden i main.go kjøres. I korte trekk er hovedpunktene følgende:
  • Gjennom DontListen: true hopper serveren over fasen for klientlytting kalt AcceptLoop.
  • Dersom alternativet InProcessServer i klientens Connect-opsjoner er aktivert, opprettes en forbindelse i minnet. Via net.Pipe lages en pipe, og den ene enden av pipen returneres til klienten som en net.Conn.
  • Klienten og serveren kommuniserer deretter via denne forbindelsen i en in-process-kommunikasjon.

Server

AcceptLoop

1// nats-server/server/server.go
2
3// Vent på klienter.
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}

• Først hoppes fasen for klientlytting, AcceptLoop, over når DontListen er true.

 1// nats-server/server/server.go
 2
 3// AcceptLoop er eksportert for enklere testing.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// Dersom vi skulle avslutte før lytteren er riktig satt opp, må vi sørge for å lukke kanalen.
 6	defer func() {
 7		if clr != nil {
 8			close(clr)
 9		}
10	}()
11
12	if s.isShuttingDown() {
13		return
14	}
15
16	// Ta et øyeblikksbilde av serveropsjonene.
17	opts := s.getOpts()
18
19	// Sett opp tilstand som gjør det mulig å stoppe serveren
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	// Varsle om at TLS er aktivert.
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	// Dersom serveren ble startet med RANDOM_PORT (-1), vil opts.Port være lik 0 i begynnelsen av denne funksjonen. Derfor må vi hente den faktiske porten
41	if opts.Port == 0 {
42		// Skriv den oppløste porten tilbake til opsjonene.
43		opts.Port = l.Addr().(*net.TCPAddr).Port
44	}
45
46	// Nå som porten er satt (dersom den var satt til RANDOM), settes serverens info Host/Port med enten verdier fra Options eller 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	// Hold oversikt over klientenes connect-URL-er. Vi kan få behov for dem senere.
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				// Signaliser at vi ikke lenger aksepterer nye klienter
61				s.ldmCh <- true
62				// Vent nå på nedstengingen...
63				<-s.quitCh
64				return true
65			}
66			return false
67		})
68	s.mu.Unlock()
69
70	// La kalleren vite at vi er klare
71	close(clr)
72	clr = nil
73}

• For ordens skyld utfører funksjonen AcceptLoop følgende trinn. Disse knytter seg til nettverkskommunikasjon, som TLS og hostPort; ved in-process-kommunikasjon er de ikke nødvendige, og de kan derfor utelates.

Client

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect forsøker å koble til NATS-systemet.
 5// URL-en kan inneholde brukernavn/passord, f.eks. nats://derek:pass@localhost:4222
 6// Kommaseparerte lister støttes også, f.eks. urlA, urlB.
 7// Opsjonene starter med standardverdiene, men kan overstyres.
 8// For å koble til websocket-porten til en NATS-server, bruk skjemaet `ws` eller `wss`, for eksempel
 9// `ws://localhost:8080`. Merk at websocket-skjemaer ikke kan kombineres 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 brukes til å opprette en tilpasset tilkobling.
 4type Options struct {
 5	// Url representerer én enkelt NATS-server-URL som klienten kobler til.
 6	// Dersom Servers-opsjonen også er satt, blir den
 7	// den første serveren i Servers-matrisen.
 8	Url string
 9
10	// InProcessServer representerer en NATS-server som kjører innenfor
11	// samme prosess. Dersom denne er satt, forsøker vi å koble til
12	// serveren direkte i stedet for å bruke eksterne TCP-forbindelser.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}

• Funksjonen Connect, som etablerer forbindelsen mellom NATS-serveren og NATS-klienten, tar imot en klient-URL og connect-opsjoner. Structen Options, som samler disse opsjonene, har et felt kalt InProcessServer av interface-typen InProcessConnProvider.

1// main.go i eksempelkoden
2
3// Initialiser ny server med alternativer
4ns, err := server.NewServer(opts)
5
6//...
7
8// Koble til serveren via en in-process-forbindelse
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

• Når nats-klienten kaller Connect og sender nats.InProcessServer(ns) til feltet InProcessServer,

1// nats-go/nats.go
2
3// InProcessServer er en Option som forsøker å etablere en forbindelse til en NATS-server som kjører innenfor prosessen, i stedet for å koble til via TCP.
4func InProcessServer(server InProcessConnProvider) Option {
5	return func(o *Options) error {
6		o.InProcessServer = server
7		return nil
8	}
9}

• erstattes opsjonens InProcessServer med den innebygde NATS-serveren,

 1// nats-go/nats.go
 2
 3// createConn kobler til serveren og pakker inn de riktige bufio-strukturene. Den håndterer riktig når en eksisterende forbindelse allerede finnes.
 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	// Dersom vi har en referanse til en in-process-server, etableres en forbindelse ved hjelp av denne.
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}

• Dette interfacet brukes i funksjonen createConn, som oppretter forbindelsen. Når opsjonen InProcessServer ikke er nil (altså gyldig), kjøres InProcesConn på InProcessServer i opsjonen,

 1// nats-server/server/server.go
 2
 3// InProcessConn returnerer en in-process-forbindelse til serveren, og unngår behovet for en TCP-lytter for lokal kommunikasjon innenfor samme prosess. Den kan brukes uavhengig av tilstanden til DontListen-opsjonen.
 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}

• som kaller InProcessConn implementert i serveren.
• Denne funksjonen kalles i nats.go, nats-klienten for Go, når InProcessServer på nc (nats-tilkoblingen) ikke er nil. Den oppretter en forbindelse (net.Conn) og binder den til serverens forbindelse.

Forbrukerdrevne interface i Go

En type implementerer et interface ved å implementere dets metoder. Det finnes ingen eksplisitt erklæring av hensikt, og ikke noe nøkkelord «implements». Implisitte interface frikobler definisjonen av et interface fra implementasjonen, som dermed kan forekomme i enhver pakke uten forhåndsavtale.

Interface implementeres implisitt, A Tour of Go

Dersom en type kun finnes for å implementere et interface, og aldri vil ha eksporterte metoder utover dette interfacet, er det ikke nødvendig å eksportere selve typen.

Generality, Effective Go

• Denne interfacedesignen illustrerer godt det Go-miljøet ofte kaller consumer defined interface (forbrukerdefinerte interface) og strukturell typing (duck typing), og derfor ønsker jeg å presentere dette temaet også.

 1// nats-go/nats.go
 2
 3// Options kan brukes til å opprette en tilpasset tilkobling.
 4type Options struct {
 5	// Url representerer én enkelt NATS-server-URL som klienten kobler til.
 6	// Dersom Servers-opsjonen også er satt, blir den
 7	// den første serveren i Servers-matrisen.
 8	Url string
 9
10	// InProcessServer representerer en NATS-server som kjører innenfor
11	// samme prosess. Dersom denne er satt, forsøker vi å koble til
12	// serveren direkte i stedet for å bruke eksterne TCP-forbindelser.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}

• La oss gå tilbake til koden. I nats.go-klienten er feltet InProcessServer i opsjons-structen definert som interfacet InProcessConnProvider, som kun utfører InProcessConn.

 1// nats-server/server/server.go
 2
 3// InProcessConn returnerer en in-process-forbindelse til serveren, og unngår behovet for en TCP-lytter for lokal kommunikasjon innenfor samme prosess. Den kan brukes uavhengig av tilstanden til DontListen-opsjonen.
 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}

• Typen som settes inn der, er imidlertid Server fra nats-server, som utfører en rekke funksjoner utover InProcessConn.
• Årsaken er at klientens eneste interesse i denne situasjonen er om objektet tilbyr InProcessConn, mens andre egenskaper i liten grad er av betydning.
• Følgelig definerer nats.go-klienten kun et forbrukerdefinert interface, InProcessConnProvider, som består av funksjonen InProcessConn() (net.Conn, error), og bruker dette.

Konklusjon

• I denne artikkelen har vi kort gjennomgått embedded mode i NATS, hvordan det fungerer, og det forbrukerdefinerte interfacet i Go, som kan observeres i NATS-koden.
• Jeg håper denne informasjonen er nyttig for dem som bruker NATS til lignende formål, og avslutter dermed artikkelen her.