---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Konfiguracja wbudowanego serwera NATS w aplikacji Go oraz struktura komunikacji w obrębie procesu
description: Omawia sposób konfiguracji i uruchamiania brokera NATS (nats-server/v2) osadzonego wewnątrz procesu Go, a także projekt komunikacji w pamięci opartej na InProcessConn oraz interfejsu definiowanego przez konsumenta.
language: pl
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Pierwsze kroki

## O NATS

> Aplikacje programowe i usługi muszą wymieniać dane. NATS to infrastruktura umożliwiająca taką wymianę danych, podzieloną na postaci wiadomości. Nazywamy ją "**message oriented middleware**" (middleware zorientowane na wiadomości).
>
> Dzięki NATS programiści aplikacji mogą:
>
> - Bez wysiłku tworzyć rozproszone i skalowalne aplikacje typu klient-serwer.
> - Przechowywać i rozprowadzać dane w czasie rzeczywistym w sposób ogólny. Można to elastycznie osiągnąć w różnych środowiskach, językach, u dostawców chmury oraz w systemach lokalnych (on-premises).
>
> [What is NATS, NATS docs](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS 是 broker wiadomości napisany w Go.

## Osadzony NATS (Embedded NATS)

> Jeśli Twoja aplikacja jest napisana w Go i jeśli odpowiada to Twojemu przypadkowi użycia oraz scenariuszom wdrożenia, możesz nawet osadzić serwer NATS wewnątrz swojej aplikacji.

[Embedding NATS, NATS docs](https://docs.nats.io/running-a-nats-service/clients#embedding-nats)

>

- Istnieje jednak pewna szczególna cecha NATS: w przypadku aplikacji napisanych w Go obsługiwany jest tryb embedded.
- Oznacza to, że zamiast powszechnego dla brokerów wiadomości modelu, w którym uruchamia się osobny serwer brokera i komunikuje się z nim klient aplikacji, sam broker można osadzić (embed) wewnątrz aplikacji napisanej w Go.

## Korzyści i przypadki użycia osadzonego NATS

- Istnieje dobrze wyjaśniający to zagadnienie [film na YouTube](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN), dlatego odsyłam do linku z filmem.
- Nawet bez wdrażania oddzielnego serwera brokera wiadomości można tworzyć modular monolith application i osiągać separate of concern, czerpiąc korzyść z możliwości osadzenia NATS w trybie embedded. Dodatkowo umożliwia to single binary deployment.
- Jest to przydatne nie tylko na platformach bez sieci (wasm), lecz także w aplikacjach offline-first.

# Przykład z oficjalnej dokumentacji

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

    // Inicjalizacja nowego serwera z opcjami
    ns, err := server.NewServer(opts)

    if err != nil {
        panic(err)
    }

    // Uruchomienie serwera w goroutine
    go ns.Start()

    // Oczekiwanie na gotowość serwera do przyjmowania połączeń
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

    // Połączenie z serwerem
    nc, err := nats.Connect(ns.ClientURL())

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Subskrypcja tematu
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Wypisanie danych wiadomości
        data := string(msg.Data)
        fmt.Println(data)

        // Zamknięcie serwera (opcjonalne)
        ns.Shutdown()
    })

    // Publikacja danych do tematu
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // Oczekiwanie na zamknięcie serwera
    ns.WaitForShutdown()
}
```

- Jest to [przykład Embedded NATS](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o) zamieszczony w oficjalnej dokumentacji NATS, jednak jeśli postępuje się zgodnie z tym przykładem, komunikacja nie odbywa się w trybie embedding.

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

- Monitorując ruch sieciowy kierowany na [localhost](http://localhost) (127.0.0.1) za pomocą polecenia `watch 'netstat -an | grep 127.0.0.1'` i uruchamiając plik go poleceniem `go run .`, można zauważyć, że pojawiają się nowe żądania sieciowe wychodzące z domyślnego portu NATS, czyli `4222`.

# Właściwe konfiguracje dla trybu embedding

- Aby komunikacja odbywała się w zamierzonym trybie embedded, potrzebne są dwie następujące opcje.
  - Klient: należy przekazać opcję `InProcessServer`.
  - Serwer: w `Server.Options` należy jawnie ustawić flagę `DontListen` na `true`.
- Te elementy nie zostały oficjalnie udokumentowane, a początki tej funkcjonalności można odnaleźć w [PR](https://github.com/nats-io/nats-server/pull/2360).

  > 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](https://github.com/nats-io/nats.go/pull/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

# Działający przykład trybu embedded

```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{
		// Konfiguracja osadzonego serwera NATS
		// ustaw DontListen na true
		DontListen: true,
	}

	// Inicjalizacja nowego serwera z opcjami
	ns, err := server.NewServer(opts)

	if err != nil {
		panic(err)
	}

	// Uruchomienie serwera w goroutine
	go ns.Start()

	// Oczekiwanie na gotowość serwera do przyjmowania połączeń
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Połączenie z serwerem przez połączenie in-process
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Subskrypcja tematu
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Wypisanie danych wiadomości
		data := string(msg.Data)
		fmt.Println(data)

		// Zamknięcie serwera (opcjonalne)
		ns.Shutdown()
	})

	// Publikacja danych do tematu
	nc.Publish(subject, []byte("Hello embedded NATS!"))

	// Oczekiwanie na zamknięcie serwera
	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

```

- Teraz widać, że nie występuje dodatkowy hop sieciowy, zgodnie z zamierzeniem.

# Pod maską

## TL;DR

![diagram1](/assets/images/embedded-nats/seq1.svg)

- Jest to diagram sekwencji pokazujący, jakie funkcje i w jaki sposób działają wewnętrznie po uruchomieniu tego kodu w `main.go`. Zasadnicze punkty są następujące.
  - Dzięki `DontListen: true` serwer pomija fazę nasłuchiwania klientów zwaną `AcceptLoop`.
  - Jeśli w opcjach połączenia klienta aktywowana jest opcja `InProcessServer`, tworzone jest połączenie in-memory: za pomocą `net.Pipe` powstaje potok, a jego drugi koniec zwracany jest klientowi jako typ `net.Conn`.
  - Klient i serwer prowadzą komunikację in-process przez to połączenie.

## Serwer

### AcceptLoop

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

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

- Najpierw, gdy `DontListen` ma wartość true, serwer pomija fazę nasłuchiwania klientów zwaną `AcceptLoop`.

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

- Dla informacji: funkcja AcceptLoop wykonuje następujące czynności. Części związane z komunikacją sieciową, takie jak `TLS` czy `hostPort`, są zbędne w przypadku komunikacji in-process, więc można je pominąć.

## Klient

### 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)
}
```

- Funkcja `Connect`, która nawiązuje połączenie między serwerem a klientem NATS, przyjmuje URL klienta oraz opcje połączenia. Struktura Options, zawierająca te opcje, posiada pole `InProcessServer` typu interfejsowego `InProcessConnProvider`.

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

- Gdy klient NATS wywołuje Connect i przekazuje `nats.InProcessServer(ns)` jako wartość pola `InProcessServer`,

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

- opcja InProcessServer zostaje zastąpiona osadzonym serwerem NATS,

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

	//...
}
```

- Ten interfejs jest wykorzystywany w funkcji `createConn`, która tworzy połączenie. Jeśli opcja `InProcessServer` nie jest nil (jest poprawna), wywoływana jest metoda `InProcessConn` z tej opcji,

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

- Wywołuje ona zaimplementowaną w serwerze funkcję `InProcessConn`.
- Ta funkcja jest wywoływana przez `nats.go`, czyli bibliotekę klienta Go, gdy opcja `InProcessServer` obiektu nc (połączenia NATS) nie jest nil. Tworzy ona połączenie (`net.Conn`) i wiąże je z połączeniem serwera.

# Interfejsy definiowane przez konsumenta w 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](https://go.dev/tour/methods/10)

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

- Ten projekt interfejsu dobrze ilustruje pojęcia, o których w Go mówi się często: interfejs definiowany przez konsumenta (consumer defined interface) oraz typowanie strukturalne (structural typing, duck typing). Dlatego chciałbym omówić także ten temat.

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

- Wróćmy do kodu. W kliencie nats.go pole opcji `InProcessServer` zostało zdefiniowane jako interfejs `InProcessConnProvider`, który obejmuje wyłącznie metodę `InProcessConn`.

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

- Jednak typem, który do niego podstawiono, jest `Server` z nats-server, który wykonuje wiele innych funkcji poza `InProcessConn`.
- Wynika to z tego, że w tej sytuacji klienta interesuje jedynie to, czy dostarczono interfejs `InProcessConn`, a pozostałe elementy nie są istotne.
- Dlatego klient nats.go używa wyłącznie interfejsu `InProcessConnProvider`, zdefiniowanego jako consumer defined interface, który zawiera jedynie funkcję `InProcessConn() (net.Conn, error)`.

# Podsumowanie

- Krótko omówiono tryb embedded NATS, sposób jego działania oraz interfejsy definiowane przez konsumenta w Go, które można zaobserwować w kodzie NATS.
- Mam nadzieję, że ta informacja okaże się pomocna dla osób korzystających z NATS w podobnym celu, i na tym kończę artykuł.