---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Структура конфигурации встроенного сервера NATS в приложении на Go и механизм межпроцессного взаимодействия в пределах одного процесса
description: В данном материале рассматриваются способ конфигурации и запуска брокера NATS (nats-server/v2), встроенного в процесс Go, а также проектирование внутриprocessной коммуникации в памяти на основе InProcessConn и интерфейса, определяемого потребителем.
language: ru
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Начало работы

## О NATS

> Программным приложениям и сервисам необходимо обмениваться данными. NATS представляет собой инфраструктуру, которая обеспечивает такой обмен данными, разделенными в виде сообщений. Мы называем это "**message oriented middleware**" (промежуточное программное обеспечение, ориентированное на сообщения).
>
> С помощью NATS разработчики приложений могут:
>
> - Без лишних усилий создавать распределенные и масштабируемые клиент-серверные приложения.
> - Хранить и распространять данные в режиме реального времени универсальным образом. Это может гибко достигаться в различных средах, языках, облачных провайдерах и локальных системах.
>
> [What is NATS, NATS docs](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS является брокером сообщений, написанным на Go.

## Встраиваемый NATS

> Если ваше приложение написано на Go и если это соответствует вашему сценарию использования и условиям развертывания, вы даже можете встроить сервер NATS непосредственно в ваше приложение.

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

>

- Также есть одна особенность NATS: для приложений, написанных на Go, поддерживается встраиваемый режим (embedded mode).
- То есть, вместо обычной для брокеров сообщений схемы, при которой отдельный сервер брокера запускается и приложение взаимодействует с ним через клиента, сам брокер можно встроить (embed) в приложение, написанное на Go.

## Преимущества и сценарии использования встраиваемого NATS

- Поскольку существует хорошо объясняющее это [видео на YouTube](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN), я ограничусь ссылкой на него.
- Даже без развертывания отдельного сервера брокера сообщений можно создать modular monolith application и добиться separate of concern, одновременно получая преимущество встраивания NATS. Кроме того, становится возможным развертывание в виде single binary.
- Помимо платформ без сети (wasm), его можно полезно применять в offline-first приложениях.

# Пример из официальной документации

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

    // Инициализация нового сервера с параметрами
    ns, err := server.NewServer(opts)

    if err != nil {
        panic(err)
    }

    // Запуск сервера через goroutine
    go ns.Start()

    // Ожидание готовности сервера к приему подключений
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

    // Подключение к серверу
    nc, err := nats.Connect(ns.ClientURL())

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Подписка на тему
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Вывод данных сообщения
        data := string(msg.Data)
        fmt.Println(data)

        // Остановка сервера (необязательно)
        ns.Shutdown()
    })

    // Публикация данных в тему
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // Ожидание завершения работы сервера
    ns.WaitForShutdown()
}
```

- Это [пример встраиваемого NATS](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o), приведенный в официальной документации NATS, однако если следовать этому примеру коду, обмен данными не будет происходить в режиме 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
```

- Если выполнить команду `watch 'netstat -an | grep 127.0.0.1'` для проверки сетевого трафика, идущего на [localhost](http://localhost) (127.0.0.1), и запустить соответствующий go-файл командой `go run .`, можно увидеть появление новых сетевых запросов, исходящих от порта NATS по умолчанию `4222`.

# Правильные конфигурации для режима embedding

- Для того чтобы обмен данными происходил в режиме embedded, необходимы следующие два параметра.
  - Client: необходимо передать опцию `InProcessServer`.
  - Server: в `Server.Options` необходимо явно указать флаг `DontListen` со значением `true`.
- Эти моменты официально не задокументированы, а истоки данной функциональности можно найти в этом [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

# Рабочий пример для режима 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{
		// для настройки встроенного сервера NATS
		// установите DonListen в true
		DontListen: true,
	}

	// Инициализация нового сервера с параметрами
	ns, err := server.NewServer(opts)

	if err != nil {
		panic(err)
	}

	// Запуск сервера через goroutine
	go ns.Start()

	// Ожидание готовности сервера к приему подключений
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Подключение к серверу через in-process соединение
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Подписка на тему
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Вывод данных сообщения
		data := string(msg.Data)
		fmt.Println(data)

		// Остановка сервера (необязательно)
		ns.Shutdown()
	})

	// Публикация данных в тему
	nc.Publish(subject, []byte("Hello embedded NATS!"))

	// Ожидание завершения работы сервера
	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

```

- Теперь можно видеть, что дополнительных сетевых переходов (network hop) не возникает, как и предполагалось.

# Под капотом

## TL;DR

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

- Это диаграмма последовательности (sequence diagram), показывающая, какие функции и как работают внутри при запуске данного кода из `main.go`. Суть можно изложить следующим образом.
  - Благодаря `DontListen: true` сервер пропускает фазу прослушивания клиентов (client listening phase) под названием `AcceptLoop`.
  - Если в опциях подключения клиента активирован `InProcessServer`, создается in-memory соединение: с помощью `net.Pipe` создается канал, а один его конец возвращается клиенту в виде `net.Conn`.
  - Клиент и сервер осуществляют in-process взаимодействие через это соединение.

## Сервер

### AcceptLoop

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

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

- Прежде всего, если `DontListen` равен true, сервер пропускает фазу прослушивания клиентов под названием `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
}
```

- Для справки: функция AcceptLoop выполняет следующие шаги. Части, связанные с сетевым взаимодействием, такие как `TLS` и `hostPort`, при in-process взаимодействии не нужны, поэтому их можно опустить.

## Клиент

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

- Функция `Connect`, которая устанавливает соединение между сервером nats и клиентом nats, принимает URL клиента и опции подключения. Структура Options, в которой собраны эти опции, содержит поле `InProcessServer` типа interface `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))
```

- Когда клиент nats выполняет Connect и передает `nats.InProcessServer(ns)` в качестве поля `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
	}
}
```

- опция InProcessServer заменяется встроенным сервером 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
	}

	//...
}
```

- Этот interface используется в функции `createConn`, которая создает соединение. Если опция `InProcessServer` не равна nil (то есть является допустимой), вызывается метод `InProcessConn` из опции InProcessServer,

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

- то есть выполняется реализованный в сервере `InProcessConn`.
- Эта функция вызывается в `nats.go`, клиенте NATS для Go, когда `InProcessServer` у nc (соединения NATS) не равен nil; она создает соединение (`net.Conn`) и привязывает его к соединению сервера.

# Interface, определяемый потребителем, в 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)

- Такой дизайн interface хорошо отражает так называемый в Go consumer defined interface (интерфейс, определяемый потребителем) и структурную типизацию (structural typing, или duck typing), поэтому я хочу рассказать и об этой теме.

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

- Вернемся к коду. В nats.go client поле структуры опций `InProcessServer` определено через interface `InProcessConnProvider`, который предусматривает только метод `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
}
```

- Однако тип, который в это поле передается, является `Server` из nats-server, и он выполняет не только InProcessConn, но и множество других функций.
- Причина в том, что в данном случае клиента интересует только то, предоставляет ли объект interface `InProcessConn`, а остальные возможности не имеют для него существенного значения.
- Поэтому nats.go client использует только consumer defined interface `InProcessConnProvider`, определяющий единственную функцию `InProcessConn() (net.Conn, error)`.

# Заключение

- Мы кратко рассмотрели режим embedded в NATS, принципы его работы, а также интерфейс, определяемый потребителем в Go, который можно проследить в коде NATS.
- Надеюсь, эта информация будет полезна тем, кто использует NATS с подобными целями, и на этом я завершаю статью.