---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Go アプリケーション内蔵型 NATS サーバーの構成とインプロセス通信の構造
description: Go プロセス内部に NATS ブローカー(nats-server/v2)を組み込んで起動する設定方法と、InProcessConn に基づくインメモリ通信およびコンシューマー定義インターフェースの設計について扱います。
language: ja
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 (Embedded NATS)

> お使いのアプリケーションが Go で書かれており、その利用ケースやデプロイメントのシナリオに適合する場合、NATS サーバーをアプリケーション内部に埋め込むことさえできます。

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

>

- そして NATS の特異な点として、Go で構成されたアプリケーションの場合、embedded mode をサポートしているということである。
- つまり、メッセージブローカーの一般的な方式である、別途ブローカーサーバーを起動し、そのサーバーとアプリケーションのクライアントを通じて通信する構造ではなく、ブローカー自体を Go で作られたアプリケーションに内蔵 (embed) できるという話である。

## 埋め込み型 NATS の利点とユースケース

- よく説明された [Youtube 動画](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN)があるため、その動画のリンクで代えさせていただく。
- 別途のメッセージブローカーサーバーをデプロイしなくても、modular monolith アプリケーションを作成して関心の分離 (separate of concern) を達成することもでき、NATS を embedded として組み込める利点を活用できる。加えて、single binary deployment も可能になる。
- platform with no network (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"

    // subject をサブスクライブする
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // メッセージデータを出力する
        data := string(msg.Data)
        fmt.Println(data)

        // サーバーをシャットダウンする (任意)
        ns.Shutdown()
    })

    // subject にデータをパブリッシュする
    nc.Publish(subject, []byte("Hello embedded NATS!"))

    // サーバーのシャットダウンを待機する
    ns.WaitForShutdown()
}
```

- NATS 公式ドキュメントに掲載されている [Embedded NATS の例](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o) であるが、そのサンプルコードどおりに進めると、embedding mode で通信が行われない。

```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 run .` でその Go ファイルを実行すると、NATS のデフォルトポートである `4222` から出発する新しいネットワークリクエストが追加されていくのを確認できる。

# embedding mode のための正しい設定

- 意図したとおり embedded mode で通信するためには、次の二つのオプションが必要である。
  - 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 mode の動作例

```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 サーバーを設定するために
		// DontListen を 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"

	// subject をサブスクライブする
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// メッセージデータを出力する
		data := string(msg.Data)
		fmt.Println(data)

		// サーバーをシャットダウンする (任意)
		ns.Shutdown()
	})

	// subject にデータをパブリッシュする
	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

```

- これで、意図したとおり追加のネットワークホップが発生しないことを確認できる。

# 内部の仕組み

## TL;DR

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

- このコードを `main.go` で実行したときに、内部でどの関数がどのように動作するかを示したシーケンス図であり、要点を説明すると以下のとおりである。
  - `DontListen: true` により、サーバーは `AcceptLoop` というクライアント待ち受けフェーズを省略する。
  - クライアントの Connect オプションのうち `InProcessServer` が有効になると、インメモリ接続を生成し、`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` のように network 通信に関係する部分であり、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)
}
```

- nats server と nats client の接続を行う `Connect` 関数は、client URL と connect Option を設定でき、それらの Option をまとめた Options struct には `InProcesConnProvider` interface 型の `InProcessServer` というフィールドが存在する。

```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 client で Connect を行う際、`InProcessServer` フィールドとして `nats.InProcessServer(ns)` を渡すと、

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

- Option の InProcessServer が embedded nats server に置き換えられ、

```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 は、connection を生成する `createConn` 関数において、`InProcessServer` Option が nil でない (有効な) 場合に、Option の InProcessServer が持つ `InProcesConn` を実行しながら、

```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 に実装された `InProcessConn` を呼び出して実行する。
- この関数は、nats の go client である `nats.go` において nc (nats connection) の `InProcessServer` が nil でない場合に呼び出され、connection (`net.Conn`) を作成し、それを server の connection に bind させる。

# Go の consumer driven interface

> 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` Option の struct フィールドは、`InProcessConn` のみを実行する `InProcessConnProvider` interface として定義されている。

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

- しかし、そこに入る型は nats-server の `Server` であり、InProcessConn だけでなく、さまざまな機能を担っている。
- なぜなら、この状況における client の関心事は、`InProcessConn` という interface を提供しているかどうかだけであり、ほかの要素はあまり重要ではないからである。
- したがって、nats.go client は `InProcessConn() (net.Conn, error)` という機能のみを定義した `InProcessConnProvider` という consumer defined interface だけを作成して利用している。

# 結論

- NATS の embedded mode とその動作方式、そして NATS のコードを通じて確認できる Go の consumer defined interface について、簡単に扱ってみた。
- この情報が、上記のような目的で NATS を使用する人々に役立つことを願い、この記事を終えたいと思う。