---
id: 8f79af5c6af89b748112aa8b77b348ab
author: prravda
title: Go-sovelluksen upotettu NATS-palvelimen kokoonpano ja prosessinsisäinen viestintärakenne
description: Tämä käsittelee menetelmää, jolla NATS-välittäjä (nats-server/v2) upotetaan ja käynnistetään Go-prosessin sisällä, sekä InProcessConn-pohjaista muistinsisäistä viestintää ja kuluttajien määrittelemien rajapintojen suunnittelua.
language: fi
date: 2025-08-04T01:05:22.150350794Z
path: /blog/posts/how-embedded-nats-communicate-with-go-application-z36089af0
category: network
---

# Aloitus

## Tietoa NATS:sta

> Ohjelmistosovellukset ja -palvelut tarvitsevat tiedon vaihtoa. NATS on infrastruktuuri, joka mahdollistaa tällaisen tiedonvaihdon viestien muodossa. Kutsumme tätä "**sanomavälitteiseksi väliohjelmistoksi**" (message oriented middleware).
>
> NATS:n avulla sovelluskehittäjät voivat:
>
> - Rakentaa hajautettuja ja skaalautuvia asiakas-palvelin-sovelluksia vaivattomasti.
> - Tallentaa ja jakaa dataa reaaliajassa yleisellä tavalla. Tämä voidaan joustavasti toteuttaa eri ympäristöissä, ohjelmointikielillä, pilvipalveluntarjoajilla ja omissa järjestelmissä.
>
> [What is NATS, NATS docs](https://docs.nats.io/nats-concepts/what-is-nats)

- NATS on Go-kielellä toteutettu viestinvälittäjä (message broker).

## Upotettu NATS

> Jos sovelluksesi on toteutettu Go-kielellä ja se sopii käyttötapaukseesi ja käyttöönottoskenaarioihisi, voit jopa upottaa NATS-palvelimen sovellukseesi.

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

>

- Lisäksi NATS:ssa on erityispiirre: Go-kielellä toteutettuja sovelluksia varten tuetaan upotettua tilaa (embedded mode).
- Toisin sanoen sen sijaan, että viestinvälittäjä käynnistettäisiin erillisenä palvelimena ja sovelluksen asiakkaat kommunikoisivat sen kautta, kuten viestinvälittäjillä yleensä tehdään, viestinvälittäjän itsensä voi upottaa (embed) Go-kielellä kirjoitettuun sovellukseen.

## Upotetun NATS:n hyödyt ja käyttötapaukset

- Aiheesta on hyvin selittävä [YouTube-video](https://youtu.be/cdTrl8UfcBo?si=KYcXOQpiyLUft6AN), joten käytän videon linkkiä sen sijaan, että selittäisin asian itse.
- Vaikka erillistä viestinvälittäjäpalvelinta ei otettaisi käyttöön, voidaan rakentaa modular monolith -sovellus ja saavuttaa separation of concerns, ja samalla hyödyntää sitä, että NATS voidaan upottaa sovellukseen. Lisäksi mahdollistuu single binary -käyttöönotto.
- Sen lisäksi, että se soveltuu alustoihin ilman verkkoyhteyttä (wasm), sitä voi hyödyntää tehokkaasti myös offline-first-sovelluksissa.

# Esimerkki virallisessa dokumentaatiossa

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

    // Alusta uusi palvelin asetuksilla
    ns, err := server.NewServer(opts)

    if err != nil {
        panic(err)
    }

    // Käynnistä palvelin goroutinena
    go ns.Start()

    // Odota, että palvelin on valmis vastaanottamaan yhteyksiä
    if !ns.ReadyForConnections(4 * time.Second) {
        panic("not ready for connection")
    }

    // Yhdistä palvelimeen
    nc, err := nats.Connect(ns.ClientURL())

    if err != nil {
        panic(err)
    }

    subject := "my-subject"

    // Tilaa aihe
    nc.Subscribe(subject, func(msg *nats.Msg) {
        // Tulosta viestin data
        data := string(msg.Data)
        fmt.Println(data)

        // Sammuta palvelin (valinnainen)
        ns.Shutdown()
    })

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

    // Odota palvelimen sammumista
    ns.WaitForShutdown()
}
```

- Tämä on NATS:n virallisen dokumentaation linkittämä [esimerkki upotetusta NATS:sta](https://dev.to/karanpratapsingh/embedding-nats-in-go-19o), mutta jos tätä esimerkkikoodia seurataan sellaisenaan, kommunikaatio ei tapahdu upotetussa tilassa (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
```

- Kun seurataan localhostin (127.0.0.1) verkkoliikennettä komennolla `watch 'netstat -an | grep 127.0.0.1'` ja suoritetaan Go-tiedosto komennolla `go run .`, nähdään, että NATS:n oletusportista `4222` lähtee uusia verkkopyyntöjä.

# Oikeat asetukset upotetulle tilalle

- Jotta kommunikaatio tapahtuisi halutulla tavalla upotetussa tilassa, tarvitaan seuraavat kaksi asetusta.
  - Asiakas: `InProcessServer`-asetus on annettava.
  - Palvelin: `Server.Options`-rakenteessa on asetettava lippu `DontListen` arvoon `true`.
- Näitä ominaisuuksia ei ole virallisesti dokumentoitu, ja tämän ominaisuuden alkuperän voi selvittää tästä [PR:stä](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

> Tämä PR lisää kolme asiaa:
>
> 1. `InProcessConn()`-funktio `Server`-tyyppiin, joka muodostaa `net.Pipe`-yhteyden NATS-palvelimeen ilman TCP-socketeja
> 2. `DontListen`-asetus, joka kertoo NATS-palvelimelle, ettei sen tule kuunnella tavanomaista TCP-kuuntelijaa
> 3. `startupComplete`-kanava, joka suljetaan juuri ennen `AcceptLoop`-silmukan käynnistämistä, ja `readyForConnections` odottaa sitä
>
> Tämän pääasiallinen motivaatio on, että meillä on sovellus, joka voi toimia joko monoliittisessa (yksiprosessisessa) tilassa tai polyliittisessa (monipro­sessisessa) tilassa. Haluaisimme voida käyttää NATS:ia molemmissa tiloissa yksinkertaisuuden vuoksi, mutta monoliittisen tilan on pystyttävä palvelemaan useita alustoja, joissa socket-yhteyksien avaaminen joko ei ole järkevää (mobiili) tai ei ole mahdollista lainkaan (WASM). Nämä muutokset mahdollistavat NATS:n käyttämisen kokonaan prosessin sisällä.
>
> Asiaan liittyvä PR [nats-io/nats.go#774](https://github.com/nats-io/nats.go/pull/774) lisää tuen asiakaspuolelle.
>
> Tämä on ensimmäinen PR tähän projektiin, joten pahoittelen etukäteen, jos olen jättänyt jotain ilmeistä huomiotta.
>
> /cc @nats-io/core

# Toimiva esimerkki upotetulle tilalle

```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{
		// Upotetun NATS-palvelimen asetukset
		// aseta DonListen arvoon true
		DontListen: true,
	}

	// Alusta uusi palvelin asetuksilla
	ns, err := server.NewServer(opts)

	if err != nil {
		panic(err)
	}

	// Käynnistä palvelin goroutinena
	go ns.Start()

	// Odota, että palvelin on valmis vastaanottamaan yhteyksiä
	if !ns.ReadyForConnections(10 * time.Second) {
		panic("not ready for connection")
	}

	// Yhdistä palvelimeen prosessinsisäisen yhteyden kautta
	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))

	if err != nil {
		panic(err)
	}

	subject := "my-subject"

	// Tilaa aihe
	nc.Subscribe(subject, func(msg *nats.Msg) {
		// Tulosta viestin data
		data := string(msg.Data)
		fmt.Println(data)

		// Sammuta palvelin (valinnainen)
		ns.Shutdown()
	})

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

	// Odota palvelimen sammumista
	ns.WaitForShutdown()
}

```

```bash
Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:37:50
                                                                     in 0.023s (0)
...ei lisälokeja

```

- Nyt nähdään, että ylimääräisiä verkkohyppyjä ei synny, kuten oli tarkoitus.

# Konepellin alla

## TL;DR

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

- Kun tämä koodi suoritetaan tiedostosta `main.go`, kyseessä on sekvenssikaavio, joka kuvaa, mitä funktioita suoritetaan sisäisesti ja missä järjestyksessä. Pääpiirteissään se on seuraava.
  - `DontListen: true` -asetuksen ansiosta palvelin ohittaa `AcceptLoop`-vaiheen, eli asiakkaiden kuunteluvaiheen.
  - Jos asiakkaan Connect-asetuksista `InProcessServer` on aktivoitu, luodaan muistinvarainen yhteys: rakennetaan `net.Pipe`-putki ja palautetaan putken toinen pää asiakkaalle `net.Conn`-tyyppinä.
  - Asiakas ja palvelin kommunikoivat tämän yhteyden kautta prosessinsisäisesti.

## Palvelin

### AcceptLoop

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

// Odota asiakkaita.
if !opts.DontListen {
	s.AcceptLoop(clientListenReady)
}
```

- Ensin, jos `DontListen` on true, palvelin ohittaa `AcceptLoop`-vaiheen, eli asiakkaiden kuunteluvaiheen.

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

// AcceptLoop on exportattu helpompaa testausta varten.
func (s *Server) AcceptLoop(clr chan struct{}) {
	// Jos poistuisimme ennen kuli kuuntelija on alustettu kunnolla,
	// varmista, että suljemme kanavan.
	defer func() {
		if clr != nil {
			close(clr)
		}
	}()

	if s.isShuttingDown() {
		return
	}

	// Ota palvelimen asetuksista kopio.
	opts := s.getOpts()

	// Alusta tila, joka mahdollistaa sammutuksen
	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)))

	// Ilmoita, että TLS on käytössä.
	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")
		}
	}

	// Jos palvelin käynnistettiin RANDOM_PORT (-1) -asetuksella, opts.Port on
	// tämän funktion alussa 0. Siksi tarvitsemme todellisen portin.
	if opts.Port == 0 {
		// Kirjoita selvitetty portti takaisin asetuksiin.
		opts.Port = l.Addr().(*net.TCPAddr).Port
	}

	// Nyt kun portti on asetettu (jos se oli RANDOM), aseta
	// palvelimen info Host/Port joko asetusten tai
	// ClientAdvertise-arvon mukaan.
	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
	}
	// Pidä kirjaa asiakkaiden yhdistys-URL-osoitteista. Saatamme tarvita niitä myöhemmin.
	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() {
				// Ilmoita, ettei uusia asiakkaita hyväksytä
				s.ldmCh <- true
				// Odota nyt sammutusta...
				<-s.quitCh
				return true
			}
			return false
		})
	s.mu.Unlock()

	// Ilmoita kutsujalle, että olemme valmiita
	close(clr)
	clr = nil
}
```

- Huomautuksena: AcceptLoop-funktio suorittaa seuraavat vaiheet. `TLS` ja `hostPort` liittyvät verkkoliikenteeseen, eikä niitä tarvita prosessinsisäisessä kommunikaatiossa, joten ne voidaan jättää pois.

## Asiakas

### InProcessServer

```go

// nats-go/nats.go

// Connect yrittää muodostaa yhteyden NATS-järjestelmään.
// URL voi sisältää käyttäjätunnuksen ja salasanan, esim. nats://derek:pass@localhost:4222.
// Pilkulla erotettuja taulukoita tuetaan myös, esim. urlA, urlB.
// Asetukset alkavat oletusarvoista, mutta niitä voidaan muuttaa.
// Yhdistääksesi NATS-palvelimen websocket-porttiin, käytä skeemaa `ws` tai `wss`, esim.
// `ws://localhost:8080`. Huomaa, että websocket-skeemoja ei voi yhdistää muihin (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-tyyppiä voidaan käyttää mukautetun yhteyden luomiseen.
type Options struct {
	// Url edustaa yhtä NATS-palvelimen URL-osoitetta, johon asiakas
	// yhdistää. Jos myös Servers-asetus on annettu, siitä tulee
	// Servers-taulukon ensimmäinen palvelin.
	Url string

	// InProcessServer edustaa samassa prosessissa käynnissä olevaa
	// NATS-palvelinta. Jos tämä on asetettu, yritämme yhdistää
	// palvelimeen suoraan sen sijaan, että käyttäisimme ulkoisia TCP-yhteyksiä.
	InProcessServer InProcessConnProvider

	//...
}
```

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

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

- `Connect`-funktio, joka yhdistää NATS-palvelimen ja NATS-asiakkaan, ottaa vastaan client-URL-osoitteen ja connect-asetuksia. Asetukset kokoava Options-rakenne sisältää kentän `InProcessServer`, jonka tyyppi on `InProcessConnProvider`-rajapinta.

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

// Alusta uusi palvelin asetuksilla
ns, err := server.NewServer(opts)

//...

// Yhdistä palvelimeen prosessinsisäisen yhteyden kautta
nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
```

- Kun NATS-asiakas suorittaa Connect-kutsun ja välittää `InProcessServer`-kentälle arvon `nats.InProcessServer(ns)`,

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

// InProcessServer on Option, joka yrittää muodostaa yhteyden samassa
// prosessissa käynnissä olevaan NATS-palvelimeen TCP-yhteyden sijaan.
func InProcessServer(server InProcessConnProvider) Option {
	return func(o *Options) error {
		o.InProcessServer = server
		return nil
	}
}
```

- asetuksen InProcessServer-kenttä korvataan upotetulla NATS-palvelimella,

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

// createConn muodostaa yhteyden palvelimeen ja kääröi sen sopiviin
// bufio-rakenteisiin. Se toimii oikein, kun olemassa oleva
// yhteys on jo käytössä.
func (nc *Conn) createConn() (err error) {
	if nc.Opts.Timeout < 0 {
		return ErrBadTimeout
	}
	if _, cur := nc.currentServer(); cur == nil {
		return ErrNoServers
	}

	// Jos meillä on viite prosessinsisäiseen palvelimeen, muodosta
	// yhteys sen kautta.
	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
	}

	//...
}
```

- Tämä rajapinta toimii yhteyden luovassa `createConn`-funktiossa. Jos `InProcessServer`-asetus ei ole nil (eli on kelvollinen), kutsutaan asetuksen InProcessServer-kentän `InProcessConn`-metodia,

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

// InProcessConn palauttaa prosessinsisäisen yhteyden palvelimeen
// ilman tarvetta TCP-kuuntelijalle paikallista yhteyttä varten
// samassa prosessissa. Tätä voidaan käyttää riippumatta
// DontListen-asetuksen tilasta.
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
}
```

- joka kutsuu palvelimelle toteutettua `InProcessConn`-metodia.
- Tämä funktio kutsutaan `nats.go`-asiakkaassa, kun nc:n (NATS-yhteyden) `InProcessServer` ei ole nil. Se luo yhteyden (`net.Conn`) ja sitoo sen palvelimen yhteyteen.

# Go-kielen kuluttajalähtöinen rajapinta

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

> Tyyppi toteuttaa rajapinnan toteuttamalla sen metodit. Aikomusta ei ilmoiteta erikseen, eikä ole "implements"-avainsanaa. Implisiittiset rajapinnat erottavat rajapinnan määrittelyn sen toteutuksesta, jolloin rajapinta voi esiintyä missä tahansa paketissa ilman ennakkosopimusta.

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

> Jos tyyppi on olemassa vain rajapinnan toteuttamista varten eikä sillä koskaan ole exportattuja metodeja tuon rajapinnan lisäksi, itse tyyppiä ei tarvitse exportata.

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

- Tämä rajapintasuunnittelu ilmentää hyvin Go-kielen usein mainitsemaa kuluttajan määrittelemää rajapintaa (consumer defined interface) ja rakenteellista tyypitystä (structural typing, eli duck typing), joten haluan esitellä myös tämän aiheen.

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

// Options-tyyppiä voidaan käyttää mukautetun yhteyden luomiseen.
type Options struct {
	// Url edustaa yhtä NATS-palvelimen URL-osoitetta, johon asiakas
	// yhdistää. Jos myös Servers-asetus on annettu, siitä tulee
	// Servers-taulukon ensimmäinen palvelin.
	Url string

	// InProcessServer edustaa samassa prosessissa käynnissä olevaa
	// NATS-palvelinta. Jos tämä on asetettu, yritämme yhdistää
	// palvelimeen suoraan sen sijaan, että käyttäisimme ulkoisia TCP-yhteyksiä.
	InProcessServer InProcessConnProvider

	//...
}
```

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

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

- Palataan vielä koodiin. NATS.go-asiakkaassa `InProcessServer`-asetuksen kenttä on määritelty `InProcessConnProvider`-rajapinnaksi, joka suorittaa vain `InProcessConn`-metodin.

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

// InProcessConn palauttaa prosessinsisäisen yhteyden palvelimeen
// ilman tarvetta TCP-kuuntelijalle paikallista yhteyttä varten
// samassa prosessissa. Tätä voidaan käyttää riippumatta
// DontListen-asetuksen tilasta.
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
}
```

- Sen tyyppi on kuitenkin nats-serverin `Server`, joka suorittaa paljon muitakin toimintoja kuin pelkän `InProcessConn`-metodin.
- Tämä johtuu siitä, että asiakkaan kannalta olennaista on vain se, tarjoaako palvelin `InProcessConn`-rajapinnan, eikä muut ominaisuudet ole merkityksellisiä.
- Siksi nats.go-asiakas käyttää vain `InProcessConnProvider`-rajapintaa, joka määrittelee pelkän `InProcessConn() (net.Conn, error)` -toiminnon. Se on kuluttajan määrittelemä rajapinta (consumer defined interface).

# Yhteenveto

- Käsittelimme lyhyesti NATS:n upotettua tilaa ja sen toimintaperiaatetta sekä Go-kielen kuluttajan määrittelemää rajapintaa, joka näkyy NATS:n koodissa.
- Toivon, että tästä tiedosta on hyötyä niille, jotka käyttävät NATS:ia samoihin tarkoituksiin, ja päätän kirjoituksen tähän.