---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Go alkalmazásba ágyazott NATS-szerver konfigurációja és folyamaton belüli (in-process) kommunikációs felépítése
description: Ez a dokumentum a NATS-brókert (nats-server/v2) egy Go-folyamaton belül beágyazva futtató konfigurációs módszert, valamint az InProcessConn alapú memórián belüli kommunikációt és a fogyasztói interfész tervezését tárgyalja.
language: hu
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# 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](https://docs.nats.io/nats-concepts/what-is-nats)

- 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](https://docs.nats.io/running-a-nats-service/clients#embedding-nats)

>

- 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ó](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN) á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

```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{}

    // Új szerver inicializálása a beállításokkal
    ns, err := server.NewServer(opts)

    if err != nil {
        panic(err)
    }

    // A szerver indítása goroutine-on keresztül
    go ns.Start()

    // Várakozás arra, hogy a szerver kész legyen a kapcsolatokra
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

    // Csatlakozás a szerverhez
    nc, err := nats.Connect(ns.ClientURL())

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Feliratkozás a tárgyra
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Az üzenet adatainak kiírása
        data := string(msg.Data)
        fmt.Println(data)

        // A szerver leállítása (opcionális)
        ns.Shutdown()
    })

    // Adatok közzététele a tárgyon
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // Várakozás a szerver leállására
    ns.WaitForShutdown()
}
```

- Ez a NATS hivatalos dokumentációban szereplő [Embedded NATS példa](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o), azonban ha ezt a példakódot követjük, a kommunikáció nem beágyazott módban történik.

```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
```

- Ha a `watch 'netstat -an | grep 127.0.0.1'` parancs segítségével figyeljük a [localhost](http://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](https://github.com/nats-io/nats-server/pull/2360) 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](https://github.com/nats-io/nats.go/pull/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

```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{
		// a beágyazott NATS szerver konfigurálásához
		// állítsa a DonListen értékét true-ra
		DontListen: true,
	}

	// Új szerver inicializálása a beállításokkal
	ns, err := server.NewServer(opts)

	if err != nil {
		panic(err)
	}

	// A szerver indítása goroutine-on keresztül
	go ns.Start()

	// Várakozás arra, hogy a szerver kész legyen a kapcsolatokra
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Csatlakozás a szerverhez folyamaton belüli kapcsolaton keresztül
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Feliratkozás a tárgyra
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Az üzenet adatainak kiírása
		data := string(msg.Data)
		fmt.Println(data)

		// A szerver leállítása (opcionális)
		ns.Shutdown()
	})

	// Adatok közzététele a tárgyon
	nc.Publish(subject, []byte("Hello embedded NATS!"))

	// Várakozás a szerver leállására
	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

```

- 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](/assets/images/embedded-nats/seq1.svg)

- 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

```go
// nats-server/server/server.go

// Wait for clients.
if !opts.DontListen {
	s.AcceptLoop(clientListenReady)
}
```

- Ha a `DontListen` értéke true, a szerver kihagyja az `AcceptLoop` nevű, kliensekre figyelő fázist.

```go
// nats-server/server/server.go

// AcceptLoop is exported for easier testing.
func (s *Server) AcceptLoop(clr chan struct{}) {
	// If we were to exit before the listener is setup properly,
	// make sure we close the channel.
	defer func() {
		if clr != nil {
			close(clr)
		}
	}()

	if s.isShuttingDown() {
		return
	}

	// Snapshot server options.
	opts := s.getOpts()

	// Setup state that can enable shutdown
	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)))

	// Alert of TLS enabled.
	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")
		}
	}

	// If server was started with RANDOM_PORT (-1), opts.Port would be equal
	// to 0 at the beginning this function. So we need to get the actual port
	if opts.Port == 0 {
		// Write resolved port back to options.
		opts.Port = l.Addr().(*net.TCPAddr).Port
	}

	// Now that port has been set (if it was set to RANDOM), set the
	// server's info Host/Port with either values from Options or
	// 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
	}
	// Keep track of client connect URLs. We may need them later.
	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() {
				// Signal that we are not accepting new clients
				s.ldmCh <- true
				// Now wait for the Shutdown...
				<-s.quitCh
				return true
			}
			return false
		})
	s.mu.Unlock()

	// Let the caller know that we are ready
	close(clr)
	clr = nil
}
```

- 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

```go

// nats-go/nats.go

// Connect will attempt to connect to the NATS system.
// The url can contain username/password semantics. e.g. nats://derek:pass@localhost:4222
// Comma separated arrays are also supported, e.g. urlA, urlB.
// Options start with the defaults but can be overridden.
// To connect to a NATS Server's websocket port, use the `ws` or `wss` scheme, such as
// `ws://localhost:8080`. Note that websocket schemes cannot be mixed with others (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 can be used to create a customized connection.
type Options struct {
	// Url represents a single NATS server url to which the client
	// will be connecting. If the Servers option is also set, it
	// then becomes the first server in the Servers array.
	Url string

	// InProcessServer represents a NATS server running within the
	// same process. If this is set then we will attempt to connect
	// to the server directly rather than using external TCP conns.
	InProcessServer InProcessConnProvider

	//...
}
```

```go
// nats-go/nats.go

type InProcessConnProvider interface {
	InProcessConn() (net.Conn, error)
}
```

- 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.

```go
// main.go of example code

// Initialize new server with options
ns, err := server.NewServer(opts)

//...

// Connect to server via in-process connection
nc, 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,

```go
// nats-go/nats.go

// InProcessServer is an Option that will try to establish a direction to a NATS server
// running within the process instead of dialing via TCP.
func InProcessServer(server InProcessConnProvider) Option {
	return func(o *Options) error {
		o.InProcessServer = server
		return nil
	}
}
```

- az opció InProcessServer mezője a beágyazott NATS szerverre cserélődik.

```go
// nats-go/nats.go

// createConn will connect to the server and wrap the appropriate
// bufio structures. It will do the right thing when an existing
// connection is in place.
func (nc *Conn) createConn() (err error) {
	if nc.Opts.Timeout < 0 {
		return ErrBadTimeout
	}
	if _, cur := nc.currentServer(); cur == nil {
		return ErrNoServers
	}

	// If we have a reference to an in-process server then establish a
	// connection using that.
	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
	}

	//...
}
```

- 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,

```go
// nats-server/server/server.go

// InProcessConn returns an in-process connection to the server,
// avoiding the need to use a TCP listener for local connectivity
// within the same process. This can be used regardless of the
// state of the DontListen option.
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
}
```

- 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](https://go.dev/tour/methods/10)

> 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](https://go.dev/doc/effective_go#generality)

- 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.

```go
// nats-go/nats.go

// Options can be used to create a customized connection.
type Options struct {
	// Url represents a single NATS server url to which the client
	// will be connecting. If the Servers option is also set, it
	// then becomes the first server in the Servers array.
	Url string

	// InProcessServer represents a NATS server running within the
	// same process. If this is set then we will attempt to connect
	// to the server directly rather than using external TCP conns.
	InProcessServer InProcessConnProvider

	//...
}
```

```go
// nats-go/nats.go

type InProcessConnProvider interface {
	InProcessConn() (net.Conn, error)
}
```

- 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.

```go
// nats-server/server/server.go

// InProcessConn returns an in-process connection to the server,
// avoiding the need to use a TCP listener for local connectivity
// within the same process. This can be used regardless of the
// state of the DontListen option.
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
}
```

- 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.