--- id: 8f79af5c6af89b748112aa8b77b348ab author: prravda title: Konfiguration av inbäddad NATS-server i Go-applikationer och arkitekturen för kommunikation inom processen description: Denna text behandlar konfigurationsmetoden för att köra en NATS-broker (nats-server/v2) inbäddad i en Go-process, samt designen av in-memory-kommunikation baserad på InProcessConn och ett gränssnitt för konsumenter. language: sv date: 2025-08-04T01:05:22.150350794Z path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0 category: network --- # 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](https://docs.nats.io/nats-concepts/what-is-nats) - 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](https://docs.nats.io/running-a-nats-service/clients#embedding-nats) > - 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](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN), 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 ```go package main import ( "fmt" "time" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" ) func main() { opts := &server.Options{} // Initialisera en ny server med options ns, err := server.NewServer(opts) if err != nil { panic(err) } // Starta servern via goroutine go ns.Start() // Vänta tills servern är redo för anslutningar if !ns.ReadyForConnections(4 * time.Second) { panic("not ready for connection") } // Anslut till servern nc, err := nats.Connect(ns.ClientURL()) if err != nil { panic(err) } subject := "my-subject" // Prenumerera på subjectet nc.Subscribe(subject, func(msg *nats.Msg) { // Skriv ut meddelandedata data := string(msg.Data) fmt.Println(data) // Stäng av servern (valfritt) ns.Shutdown() }) // Publicera data till subjectet nc.Publish(subject, []byte("Hello embedded NATS!")) // Vänta på att servern stängs av ns.WaitForShutdown() } ``` - Detta är [exemplet på Embedded NATS](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o) som NATS officiella dokumentation länkar till, men om man följer exempelkoden så sker kommunikationen inte i embedding-läge. ```bash Every 2.0s: netstat -an | grep 127.0.0.1 pravdalaptop-home.local: 02:34:20 in 0.017s (0) ... tcp4 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](http://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](https://github.com/nats-io/nats-server/pull/2360). > 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](https://github.com/nats-io/nats.go/pull/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 ```go package main import ( "fmt" "time" "github.com/nats-io/nats-server/v2/server" "github.com/nats-io/nats.go" ) func main() { opts := &server.Options{ // för att konfigurera den inbäddade NATS-servern // sätt DonListen till true DontListen: true, } // Initialisera en ny server med options ns, err := server.NewServer(opts) if err != nil { panic(err) } // Starta servern via goroutine go ns.Start() // Vänta tills servern är redo för anslutningar if !ns.ReadyForConnections(10 * time.Second) { panic("not ready for connection") } // Anslut till servern via in-process-anslutning nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns)) if err != nil { panic(err) } subject := "my-subject" // Prenumerera på subjectet nc.Subscribe(subject, func(msg *nats.Msg) { // Skriv ut meddelandedata data := string(msg.Data) fmt.Println(data) // Stäng av servern (valfritt) ns.Shutdown() }) // Publicera data till subjectet nc.Publish(subject, []byte("Hello embedded NATS!")) // Vänta på att servern stängs av ns.WaitForShutdown() } ``` ```bash Every 2.0s: netstat -an | grep 127.0.0.1 pravdalaptop-home.local: 02:37:50 in 0.023s (0) ...no additional logs ``` - Nu kan man se att inga extra nätverkshopp uppstår, precis som avsett. # Under huven ## TL;DR ![diagram1](/assets/images/embedded-nats/seq1.svg) - 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 ```go // nats-server/server/server.go // Vänta på klienter. if !opts.DontListen { s.AcceptLoop(clientListenReady) } ``` - Om `DontListen` är true hoppar servern först över client-lyssningsfasen `AcceptLoop`. ```go // nats-server/server/server.go // AcceptLoop exporteras för enklare testning. func (s *Server) AcceptLoop(clr chan struct{}) { // Om vi skulle avsluta innan lyssnaren är korrekt konfigurerad, // se till att kanalen stängs. defer func() { if clr != nil { close(clr) } }() if s.isShuttingDown() { return } // Ta en ögonblicksbild av serverns options. opts := s.getOpts() // Konfigurera tillstånd som gör det möjligt att stänga av s.mu.Lock() hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port)) l, e := natsListen("tcp", hp) s.listenerErr = e if e != nil { s.mu.Unlock() s.Fatalf("Error listening on port: %s, %q", hp, e) return } s.Noticef("Listening for client connections on %s", net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port))) // Meddela att TLS är aktiverat. if opts.TLSConfig != nil { s.Noticef("TLS required for client connections") if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 { s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect") } } // Om servern startades med RANDOM_PORT (-1) är opts.Port lika med // 0 i början av denna funktion. Därför måste vi hämta den faktiska porten if opts.Port == 0 { // Skriv tillbaka den upplösta porten till options. opts.Port = l.Addr().(*net.TCPAddr).Port } // Nu när porten har satts (om den var RANDOM), sätt // serverns info Host/Port med antingen värden från Options eller // ClientAdvertise. if err := s.setInfoHostPort(); err != nil { s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err) l.Close() s.mu.Unlock() return } // Håll reda på klientanslutnings-URL:erna. Vi kan behöva dem senare. s.clientConnectURLs = s.getClientConnectURLs() s.listener = l go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) }, func(_ error) bool { if s.isLameDuckMode() { // Signalera att vi inte längre accepterar nya klienter s.ldmCh <- true // Vänta nu på Shutdown... <-s.quitCh return true } return false }) s.mu.Unlock() // Låt anroparen veta att vi är redo close(clr) clr = nil } ``` - 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 ```go // nats-go/nats.go // Connect försöker ansluta till NATS-systemet. // URL:en kan innehålla användarnamn/lösenord, t.ex. nats://derek:pass@localhost:4222. // Kommaseparerade arrayer stöds också, t.ex. urlA, urlB. // Options börjar med standardvärden men kan åsidosättas. // För att ansluta till en NATS-servers websocket-port, använd schemat `ws` eller `wss`, t.ex. // `ws://localhost:8080`. Observera att websocket-scheman inte kan blandas med andra (nats/tls). func Connect(url string, options ...Option) (*Conn, error) { opts := GetDefaultOptions() opts.Servers = processUrlString(url) for _, opt := range options { if opt != nil { if err := opt(&opts); err != nil { return nil, err } } } return opts.Connect() } ``` ```go // nats-go/nats.go // Options kan användas för att skapa en anpassad anslutning. type Options struct { // Url representerar en enskild NATS-server-URL till vilken klienten // ansluter. Om Servers-optionen också är satt blir den // första servern i Servers-arrayen. Url string // InProcessServer representerar en NATS-server som körs inom // samma process. Om denna är satt försöker vi ansluta // direkt till servern i stället för att använda externa TCP-anslutningar. InProcessServer InProcessConnProvider //... } ``` ```go // nats-go/nats.go type InProcessConnProvider interface { InProcessConn() (net.Conn, error) } ``` - 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`. ```go // main.go i exempelkoden // Initialisera en ny server med options ns, err := server.NewServer(opts) //... // Anslut till servern via in-process-anslutning nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns)) ``` - När NATS-klienten anropar Connect och skickar `nats.InProcessServer(ns)` till fältet `InProcessServer` ```go // nats-go/nats.go // InProcessServer är en Option som försöker etablera en riktning till en NATS-server // som körs inom processen i stället för att ringa via TCP. func InProcessServer(server InProcessConnProvider) Option { return func(o *Options) error { o.InProcessServer = server return nil } } ``` - ersätts option-fältet InProcessServer med den inbäddade NATS-servern, ```go // nats-go/nats.go // createConn ansluter till servern och omsluter lämpliga // bufio-strukturer. Den hanterar rätt när en befintlig // anslutning redan finns. func (nc *Conn) createConn() (err error) { if nc.Opts.Timeout < 0 { return ErrBadTimeout } if _, cur := nc.currentServer(); cur == nil { return ErrNoServers } // Om vi har en referens till en in-process-server, etablera en // anslutning med den. if nc.Opts.InProcessServer != nil { conn, err := nc.Opts.InProcessServer.InProcessConn() if err != nil { return fmt.Errorf("failed to get in-process connection: %w", err) } nc.conn = conn nc.bindToNewConn() return nil } //... } ``` - 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. ```go // nats-server/server/server.go // InProcessConn returnerar en in-process-anslutning till servern, // vilket undviker behovet av en TCP-listener för lokal anslutning // inom samma process. Detta kan användas oavsett // tillståndet för optionen DontListen. func (s *Server) InProcessConn() (net.Conn, error) { pl, pr := net.Pipe() if !s.startGoRoutine(func() { s.createClientInProcess(pl) s.grWG.Done() }) { pl.Close() pr.Close() return nil, fmt.Errorf("failed to create connection") } return pr, nil } ``` - 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](https://go.dev/tour/methods/10) > 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](https://go.dev/doc/effective_go#generality) - 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. ```go // nats-go/nats.go // Options kan användas för att skapa en anpassad anslutning. type Options struct { // Url representerar en enskild NATS-server-URL till vilken klienten // ansluter. Om Servers-optionen också är satt blir den // första servern i Servers-arrayen. Url string // InProcessServer representerar en NATS-server som körs inom // samma process. Om denna är satt försöker vi ansluta // direkt till servern i stället för att använda externa TCP-anslutningar. InProcessServer InProcessConnProvider //... } ``` ```go // nats-go/nats.go type InProcessConnProvider interface { InProcessConn() (net.Conn, error) } ``` - 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`. ```go // nats-server/server/server.go // InProcessConn returnerar en in-process-anslutning till servern, // vilket undviker behovet av en TCP-listener för lokal anslutning // inom samma process. Detta kan användas oavsett // tillståndet för optionen DontListen. func (s *Server) InProcessConn() (net.Conn, error) { pl, pr := net.Pipe() if !s.startGoRoutine(func() { s.createClientInProcess(pl) s.grWG.Done() }) { pl.Close() pr.Close() return nil, fmt.Errorf("failed to create connection") } return pr, nil } ``` - 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.