GoSuda

Go alkalmazásba ágyazott NATS-szerver konfigurációja és folyamaton belüli (in-process) kommunikációs felépítése

By prravda views ...

Első lépések

A NATS-ról

A szoftveralkalmazásoknak és szolgáltatásoknak adatokat kell cserélniük. A NATS egy olyan infrastruktúra, amely lehetővé teszi az ilyen adatcserét, üzenetek formájában szegmentálva. Ezt "message oriented middleware"-nek (üzenetorientált köztesrétegnek) nevezzük.

A NATS segítségével az alkalmazásfejlesztők:

  • Könnyedén építhetnek elosztott és skálázható kliens-szerver alkalmazásokat.
  • Adatokat tárolhatnak és terjeszthetnek valós időben, általános módon. Ez rugalmasan megvalósítható különböző környezetekben, programozási nyelveken, felhőszolgáltatóknál és helyben üzemeltetett rendszerekben.

What is NATS, NATS docs

  • A NATS egy Go nyelven írt üzenetközvetítő (message broker).

Beágyazott NATS

Ha az alkalmazásod Go nyelven készült, és ez illeszkedik a használati eseteidhez és telepítési forgatókönyveidhez, akár a NATS szervert is beágyazhatod az alkalmazásodba.

Embedding NATS, NATS docs

  • A NATS egyik sajátossága, hogy a Go nyelven írt alkalmazások esetében támogatja a beágyazott (embedded) módot.
  • Vagyis a szokásos üzenetközvetítői megközelítéssel szemben, ahol külön brókerszervert indítunk, és az alkalmazás kliensén keresztül kommunikálunk vele, maga a bróker is beágyazható abba a Go nyelven írt alkalmazásba.

A beágyazott NATS előnyei és használati esetei

  • A jól kidolgozott YouTube-videó áll rendelkezésre, ezért a témát a videó linkjével intézem el.
  • Anélkül, hogy külön üzenetközvetítő szervert telepítenénk, moduláris monolit (modular monolith) alkalmazást hozhatunk létre, és így a felelősségek szétválasztását (separation of concerns) is elérhetjük, miközben kihasználhatjuk azt az előnyt, hogy a NATS beágyazható. Emellett egyetlen bináris fájlként történő telepítés (single binary deployment) is lehetővé válik.
  • Hálózat nélküli platformokon (wasm) és offline-first alkalmazásokban is hasznosan alkalmazható.

Példa a hivatalos dokumentációból

 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    // Új szerver inicializálása a beállításokkal
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // A szerver indítása goroutine-on keresztül
22    go ns.Start()
23
24    // Várakozás arra, hogy a szerver kész legyen a kapcsolatokra
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Csatlakozás a szerverhez
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Feliratkozás a tárgyra
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Az üzenet adatainak kiírása
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // A szerver leállítása (opcionális)
45        ns.Shutdown()
46    })
47
48    // Adatok közzététele a tárgyon
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Várakozás a szerver leállására
52    ns.WaitForShutdown()
53}
  • Ez a NATS hivatalos dokumentációban szereplő Embedded NATS példa, azonban ha ezt a példakódot követjük, a kommunikáció nem beágyazott módban történik.
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
  • Ha a watch 'netstat -an | grep 127.0.0.1' parancs segítségével figyeljük a localhost (127.0.0.1) felé és onnan irányuló hálózati forgalmat, miközben a go run . paranccsal futtatjuk a Go fájlt, látható, hogy a NATS alapértelmezett portjáról, a 4222-ről induló új hálózati kérések jelennek meg.

Helyes konfigurációk a beágyazott módhoz

  • Ahhoz, hogy a kommunikáció a kívánt módon, beágyazott módban történjen, a következő két beállításra van szükség:

    • Kliens: az InProcessServer opciót kell megadni.
    • Szerver: a Server.Options struktúrában a DontListen jelzőt true értékre kell állítani.

  • Ezek a részletek hivatalosan nincsenek dokumentálva, a funkció kezdetei ebben a PR-ben követhetők nyomon.

    Ez a PR három dolgot ad hozzá:

    1. Az InProcessConn() függvényt a Server-hez, amely net.Pipe segítségével kapcsolatot hoz létre a NATS szerverrel TCP socketek használata nélkül
    2. A DontListen opciót, amely utasítja a NATS szervert, hogy ne figyeljen a szokásos TCP listeneren
    3. A startupComplete csatornát, amely közvetlenül az AcceptLoop indítása előtt zárul le, a readyForConnections pedig erre vár

    Ennek fő motivációja, hogy van egy alkalmazásunk, amely futhat monolit (egyfolyamatos) módban vagy polilit (többfolyamatos) módban. Az egyszerűség kedvéért mindkét módban szeretnénk NATS-t használni, de a monolit módnak számos olyan platformot kell kiszolgálnia, ahol socket-kapcsolatok nyitása vagy értelmetlen (mobil) vagy egyszerűen nem lehetséges (WASM). Ezek a változtatások lehetővé teszik, hogy a NATS-t teljesen folyamaton belül használjuk.

    Egy kísérő PR, a nats-io/nats.go#774, támogatást ad a kliens oldalhoz.

    Ez az első PR-em ebbe a projektbe, ezért előre elnézést kérek, ha valami nyilvánvalót kihagytam.

    /cc @nats-io/core

Működő példa a beágyazott módhoz

 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		// a beágyazott NATS szerver konfigurálásához
14		// állítsa a DonListen értékét true-ra
15		DontListen: true,
16	}
17
18	// Új szerver inicializálása a beállításokkal
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// A szerver indítása goroutine-on keresztül
26	go ns.Start()
27
28	// Várakozás arra, hogy a szerver kész legyen a kapcsolatokra
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Csatlakozás a szerverhez folyamaton belüli kapcsolaton keresztül
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	// Feliratkozás a tárgyra
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Az üzenet adatainak kiírása
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// A szerver leállítása (opcionális)
49		ns.Shutdown()
50	})
51
52	// Adatok közzététele a tárgyon
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Várakozás a szerver leállására
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
  • Most már látható, hogy a kívánt módon nem keletkezik további hálózati ugrás (network hop).

A motorháztető alatt

TL;DR

diagram1

  • Ez a szekvenciadiagram azt mutatja be, hogy a kód main.go fájlban történő futtatásakor mely függvények, hogyan működnek belsőleg. A lényeg az alábbi:
    • A DontListen: true beállítás hatására a szerver kihagyja az AcceptLoop nevű, kliensekre figyelő fázist.
    • Ha a kliens Connect opciói között az InProcessServer aktiválódik, a rendszer memórián belüli kapcsolatot hoz létre, a net.Pipe segítségével csövet (pipe) készít, majd a cső egyik végét net.Conn típusként visszaadja a kliensnek.
    • A kliens és a szerver ezen a kapcsolaton keresztül folyamaton belüli kommunikációt folytat.

Szerver

AcceptLoop

1// nats-server/server/server.go
2
3// Wait for clients.
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}
  • Ha a DontListen értéke true, a szerver kihagyja az AcceptLoop nevű, kliensekre figyelő fázist.
 1// nats-server/server/server.go
 2
 3// AcceptLoop is exported for easier testing.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// If we were to exit before the listener is setup properly,
 6	// make sure we close the channel.
 7	defer func() {
 8		if clr != nil {
 9			close(clr)
10		}
11	}()
12
13	if s.isShuttingDown() {
14		return
15	}
16
17	// Snapshot server options.
18	opts := s.getOpts()
19
20	// Setup state that can enable 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	// Alert of TLS enabled.
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	// If server was started with RANDOM_PORT (-1), opts.Port would be equal
42	// to 0 at the beginning this function. So we need to get the actual port
43	if opts.Port == 0 {
44		// Write resolved port back to options.
45		opts.Port = l.Addr().(*net.TCPAddr).Port
46	}
47
48	// Now that port has been set (if it was set to RANDOM), set the
49	// server's info Host/Port with either values from Options or
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	// Keep track of client connect URLs. We may need them later.
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 that we are not accepting new clients
65				s.ldmCh <- true
66				// Now wait for the Shutdown...
67				<-s.quitCh
68				return true
69			}
70			return false
71		})
72	s.mu.Unlock()
73
74	// Let the caller know that we are ready
75	close(clr)
76	clr = nil
77}
  • Az AcceptLoop függvény a következő lépéseket hajtja végre. A TLS és a hostPort kezelése a hálózati kommunikációhoz kapcsolódik, folyamaton belüli kommunikáció esetén ezekre nincs szükség, így ezek a részek elhagyhatók.

Kliens

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect will attempt to connect to the NATS system.
 5// The url can contain username/password semantics. e.g. nats://derek:pass@localhost:4222
 6// Comma separated arrays are also supported, e.g. urlA, urlB.
 7// Options start with the defaults but can be overridden.
 8// To connect to a NATS Server's websocket port, use the `ws` or `wss` scheme, such as
 9// `ws://localhost:8080`. Note that websocket schemes cannot be mixed with others (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 can be used to create a customized connection.
 4type Options struct {
 5	// Url represents a single NATS server url to which the client
 6	// will be connecting. If the Servers option is also set, it
 7	// then becomes the first server in the Servers array.
 8	Url string
 9
10	// InProcessServer represents a NATS server running within the
11	// same process. If this is set then we will attempt to connect
12	// to the server directly rather than using external TCP conns.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • A NATS szerver és a NATS kliens közötti kapcsolatot létrehozó Connect függvény a kliens URL-jét és a connect opciókat fogadja, a Options struktúra, amely ezeket az opciókat tartalmazza, pedig egy InProcessServer nevű mezőt tartalmaz, amelynek típusa az InProcessConnProvider interfész.
1// main.go of example code
2
3// Initialize new server with options
4ns, err := server.NewServer(opts)
5
6//...
7
8// Connect to server via in-process connection
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
  • Amikor a NATS kliens a Connect hívást végzi, és az InProcessServer mezőnek a nats.InProcessServer(ns) értéket adjuk át,
 1// nats-go/nats.go
 2
 3// InProcessServer is an Option that will try to establish a direction to a NATS server
 4// running within the process instead of dialing via TCP.
 5func InProcessServer(server InProcessConnProvider) Option {
 6	return func(o *Options) error {
 7		o.InProcessServer = server
 8		return nil
 9	}
10}
  • az opció InProcessServer mezője a beágyazott NATS szerverre cserélődik.
 1// nats-go/nats.go
 2
 3// createConn will connect to the server and wrap the appropriate
 4// bufio structures. It will do the right thing when an existing
 5// connection is in place.
 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	// If we have a reference to an in-process server then establish a
15	// connection using that.
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}
  • Ez az interfész a createConn függvényben jelenik meg, amely a kapcsolatot hozza létre. Ha az InProcessServer opció nem nil (vagyis érvényes), az opcióban szereplő InProcessServer InProcessConn metódusa fut le,
 1// nats-server/server/server.go
 2
 3// InProcessConn returns an in-process connection to the server,
 4// avoiding the need to use a TCP listener for local connectivity
 5// within the same process. This can be used regardless of the
 6// state of the DontListen option.
 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}
  • A szerveren implementált InProcessConn függvényt hívja meg.
  • Ezt a függvényt a NATS Go kliense, a nats.go hívja meg, amikor a nc (NATS kapcsolat) InProcessServer értéke nem nil. Ez létrehozza a kapcsolatot (net.Conn), majd ezt a szerver kapcsolatához köti.

A Go fogyasztó által definiált interfészei

Egy típus úgy valósít meg egy interfészt, hogy implementálja annak metódusait. Nincs explicit szándéknyilatkozat, nincs "implements" kulcsszó. Az implicit interfészek szétválasztják az interfész definícióját annak megvalósításától, amely így bármely csomagban megjelenhet, előzetes egyeztetés nélkül.

Interfaces are implemented implicitly, A Tour of Go

Ha egy típus kizárólag egy interfész megvalósítására létezik, és az interfészen túl soha nem rendelkezik exportált metódusokkal, nincs szükség a típus magának az exportálására.

Generality, Effective Go

  • Ez az interfésztervezés jól példázza a Go nyelvben gyakran emlegetett, fogyasztó által definiált interfészt (consumer defined interface) és a strukturális típusosságot (structural typing, duck typing), ezért ezt a témát is szeretném bemutatni.
 1// nats-go/nats.go
 2
 3// Options can be used to create a customized connection.
 4type Options struct {
 5	// Url represents a single NATS server url to which the client
 6	// will be connecting. If the Servers option is also set, it
 7	// then becomes the first server in the Servers array.
 8	Url string
 9
10	// InProcessServer represents a NATS server running within the
11	// same process. If this is set then we will attempt to connect
12	// to the server directly rather than using external TCP conns.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Térjünk vissza a kódhoz. A nats.go kliensben az InProcessServer opciós struktúra mezője az InProcessConnProvider interfészként van definiálva, amely csak az InProcessConn műveletet tartalmazza.
 1// nats-server/server/server.go
 2
 3// InProcessConn returns an in-process connection to the server,
 4// avoiding the need to use a TCP listener for local connectivity
 5// within the same process. This can be used regardless of the
 6// state of the DontListen option.
 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}
  • Ennek az interfésznek a megvalósítója viszont a nats-server Server típusa, amely az InProcessConn mellett számos más funkciót is ellát.
  • Ennek oka, hogy ebben a helyzetben a kliens számára csak az számít, hogy az InProcessConn interfészt biztosítják-e, a többi részlet nem lényeges.
  • Ezért a nats.go kliens csak az InProcessConn() (net.Conn, error) műveletet meghatározó InProcessConnProvider nevű, fogyasztó által definiált interfészt hozta létre és használja.

Összegzés

  • Röviden áttekintettük a NATS beágyazott módját és működését, valamint a NATS kódján keresztül megfigyelhető, Go nyelvben használt fogyasztó által definiált interfészt.
  • Remélem, a leírás hasznos lesz azok számára, akik hasonló célból használják a NATS-t, ezzel a cikket be is fejezem.