---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Configurarea serverului NATS încorporat într-o aplicație Go și structura comunicării în proces
description: Acest document prezintă metoda de configurare a unui broker NATS (nats-server/v2) integrat și rulat în interiorul unui proces Go, precum și proiectarea comunicării în memorie bazate pe InProcessConn și a interfeței de consumator definite.
language: ro
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Noțiuni de bază

## Despre NATS

> Aplicațiile și serviciile software trebuie să schimbe date. NATS este o infrastructură care permite acest schimb de date, segmentat sub formă de mesaje. Numim acest lucru "**message oriented middleware**".
>
> Cu NATS, dezvoltatorii de aplicații pot:
>
> - Să construiască aplicații client-server distribuite și scalabile fără efort.
> - Să stocheze și să distribuie date în timp real, într-un mod general. Acest lucru poate fi realizat flexibil în diverse medii, limbaje, furnizori de cloud și sisteme on-premises.
>
> [What is NATS, NATS docs](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS 是一个 Go 语言编写的消息代理。

## NATS încorporat

> Dacă aplicația dumneavoastră este scrisă în Go și dacă se potrivește cazului de utilizare și scenariilor de implementare, puteți chiar să încorporați un server NATS în interiorul aplicației.

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

>

- Și există o particularitate a NATS: pentru aplicațiile construite în Go, este suportat modul embedded.
- Cu alte cuvinte, în loc de structura obișnuită a unui broker de mesaje, în care un server de broker separat este pornit și comunicarea se realizează prin clienții aplicației cu acel server, brokerul propriu-zis poate fi încorporat (embed) în aplicația construită în Go.

## Beneficiile și cazurile de utilizare ale NATS încorporat

- Există un [videoclip YouTube](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN) bine explicat, astfel că il substituim cu linkul videoclipului.
- Chiar și fără implementarea unui server de broker de mesaje separat, se poate crea o aplicație modular monolith și se pot realiza separate of concern, valorificând avantajul încorporării NATS ca embedded. În plus, devine posibilă și implementarea sub forma unui single binary.
- Pe lângă platformele fără rețea (wasm), poate fi utilizat cu folos și în aplicații offline-first.

# Exemplu din documentația oficială

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

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

    if err != nil {
        panic(err)
    }

    // Start the server via goroutine
    go ns.Start()

    // Wait for server to be ready for connections
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

    // Connect to server
    nc, err := nats.Connect(ns.ClientURL())

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Subscribe to the subject
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Print message data
        data := string(msg.Data)
        fmt.Println(data)

        // Shutdown the server (optional)
        ns.Shutdown()
    })

    // Publish data to the subject
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // Wait for server shutdown
    ns.WaitForShutdown()
}
```

- Acesta este [exemplul Embedded NATS](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o) prezentat în documentația oficială NATS, însă, dacă se urmează acest cod de exemplu, comunicarea nu se realizează în modul 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
```

- Prin comanda `watch 'netstat -an | grep 127.0.0.1'`, care verifică rețeaua care circulă către [localhost](http://localhost) (127.0.0.1), și prin rularea fișierului Go cu `go run .`, se poate observa că apar noi cereri de rețea care pornesc de la portul implicit al NATS, `4222`.

# Configurațiile corecte pentru modul embedding

- Pentru a comunica în modul embedded așa cum se dorește, sunt necesare două opțiuni:
  - Client: trebuie transmisă opțiunea `InProcessServer`.
  - Server: în `Server.Options` trebuie specificat explicit flag-ul `DontListen` cu valoarea `true`.
- Aceste aspecte nu au fost documentate oficial, iar originea acestei funcționalități poate fi urmărită în acest [PR](https://github.com/nats-io/nats-server/pull/2360).

  > Acest PR adaugă trei lucruri:
  >
  > 1. Funcția `InProcessConn()` pentru `Server`, care construiește un `net.Pipe` pentru a obține o conexiune la serverul NATS fără a utiliza socket-uri TCP
  > 2. Opțiunea `DontListen`, care îi spune serverului NATS să nu asculte pe listener-ul TCP obișnuit
  > 3. Canalul `startupComplete`, care este închis chiar înainte de a începe `AcceptLoop`, iar `readyForConnections` va aștepta după el
  >
  > Motivația principală este că avem o aplicație care poate rula fie în modul monolith (single-process), fie în modul polylith (multi-process). Dorim să putem folosi NATS în ambele moduri, pentru simplitate, însă modul monolith trebuie să poată deservi o varietate de platforme unde deschiderea conexiunilor prin socket nu are sens (mobile) sau pur și simplu nu este posibilă (WASM). Aceste modificări ne vor permite să utilizăm NATS integral in-process.
  >
  > Un PR însoțitor [nats-io/nats.go#774](https://github.com/nats-io/nats.go/pull/774) adaugă suport pe partea de client.
  >
  > Acesta este primul meu PR la acest proiect, așa că îmi cer scuze din timp dacă am omis ceva evident.
  >
  > /cc @nats-io/core

# Exemplu funcțional pentru modul 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{
		// for configuring the embeded NATS server
		// set DonListen as true
		DontListen: true,
	}

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

	if err != nil {
		panic(err)
	}

	// Start the server via goroutine
	go ns.Start()

	// Wait for server to be ready for connections
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Connect to server via in-process connection
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Subscribe to the subject
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Print message data
		data := string(msg.Data)
		fmt.Println(data)

		// Shutdown the server (optional)
		ns.Shutdown()
	})

	// Publish data to the subject
	nc.Publish(subject, []byte("Hello embedded NATS!"))

	// Wait for server shutdown
	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

```

- Acum se poate observa că nu mai apare niciun salt de rețea suplimentar, cum era intenționat.

# În profunzime

## TL;DR

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

- Aceasta este diagrama de tip sequence care arată ce funcții sunt apelate intern și cum funcționează atunci când acest cod este executat din `main.go`. Esența este următoarea:
  - Prin `DontListen: true`, serverul omite faza de ascultare a clienților numită `AcceptLoop`.
  - Dacă opțiunea `InProcessServer` a Connect-ului clientului este activată, se creează o conexiune in-memory, se construiește un pipe prin `net.Pipe`, iar capătul pipe-ului este returnat clientului ca tip `net.Conn`.
  - Clientul și serverul realizează comunicarea in-process prin această conexiune.

## Server

### AcceptLoop

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

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

- Mai întâi, dacă `DontListen` este true, se omite faza de ascultare a clienților numită `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
}
```

- Ca observație, funcția AcceptLoop parcurge următorii pași. Părțile legate de comunicarea în rețea, cum sunt `TLS` sau `hostPort`, nu sunt necesare în cazul comunicării in-process, așadar pot fi omise fără probleme.

## Client

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

- Funcția `Connect`, care realizează conectarea dintre serverul NATS și clientul NATS, primește URL-ul clientului și opțiunile de conectare, iar structura Options, în care sunt adunate aceste opțiuni, conține un câmp numit `InProcessServer` de tipul interfață `InProcesConnProvider`.

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

- Când clientul NATS apelează Connect și transmite `nats.InProcessServer(ns)` ca valoare a câmpului `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
	}
}
```

- opțiunea InProcessServer este înlocuită cu serverul NATS încorporat,

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

	//...
}
```

- Această interfață este utilizată în funcția `createConn`, care creează conexiunea. Dacă opțiunea `InProcessServer` nu este nil (adică este validă), se execută `InProcessConn` al lui InProcessServer din opțiune,

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

- Se apelează `InProcessConn` implementată în server.
- Această funcție este apelată din `nats.go`, clientul Go pentru NATS, atunci când `InProcessServer` al nc (conexiunii NATS) nu este nil; ea creează conexiunea (`net.Conn`), care este apoi legată de conexiunea serverului.

# Interfață definită de consumator în Go

> Un tip implementează o interfață prin simpla implementare a metodelor sale. Nu există o declarație explicită a intenției, nu există cuvântul cheie "implements". Interfețele implicite decuplează definiția unei interfețe de implementarea sa, care poate apărea apoi în orice pachet, fără aranjamente prealabile.

[Interfaces are implemented implicitly, A Tour of Go](https://go.dev/tour/methods/10)

> Dacă un tip există doar pentru a implementa o interfață și nu va avea niciodată metode exportate dincolo de acea interfață, nu este necesar să exportăm tipul în sine.

[Generality, Effective Go](https://go.dev/doc/effective_go#generality)

- Această abordare a interfețelor reflectă bine ceea ce în Go se numește interfață definită de consumator și tipizare structurală (duck typing), astfel că merită să prezentăm și acest subiect.

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

- Revenind la cod: în clientul nats.go, câmpul de structură al opțiunii `InProcessServer` este definit ca interfața `InProcessConnProvider`, care realizează doar `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
}
```

- Totuși, tipul care este transmis acolo este `Server` din nats-server, care îndeplinește nu doar InProcessConn, ci și multe alte funcții.
- Acest lucru se explică prin faptul că, în acest context, interesul clientului este doar dacă a fost furnizată interfața `InProcessConn` sau nu, iar celelalte aspecte nu sunt deosebit de relevante.
- Prin urmare, nats.go client folosește doar interfața `InProcessConnProvider`, o interfață definită de consumator care specifică doar funcționalitatea `InProcessConn() (net.Conn, error)`.

# Concluzie

- Am tratat pe scurt modul embedded al NATS, modul său de funcționare și interfața definită de consumator din Go, care poate fi observată în codul NATS.
- Sper că aceste informații vor fi utile celor care utilizează NATS în scopurile menționate mai sus, și închei aici acest articol.