GoSuda

Go Uygulamasına Gömülü NATS Sunucusu Yapılandırması ve Süreç İçi İletişim Mimarisi

By prravda views ...

Başlarken

NATS Hakkında

Yazılım uygulamaları ve servisleri verilerini birbiriyle alışverişe ihtiyaç duyar. NATS, bu tür veri alışverişini mesajlar biçiminde parçalara ayırarak mümkün kılan bir altyapıdır. Buna "message oriented middleware" (mesaj odaklı ara katman yazılımı) diyoruz.

NATS ile uygulama geliştiricileri:

  • Dağıtık ve ölçeklenebilir istemci-sunucu uygulamalarını zahmetsizce geliştirebilir.
  • Verileri gerçek zamanlı olarak genel bir biçimde saklayabilir ve dağıtabilir. Bu, çeşitli ortamlar, diller, bulut sağlayıcıları ve yerinde (on-premises) sistemler genelinde esnek bir şekilde gerçekleştirilebilir.

What is NATS, NATS docs

  • NATS, Go ile yapılandırılmış bir mesaj brokeridır.

Gömülü NATS (Embedded NATS)

Uygulamanız Go ile yazılmışsa ve kullanım senaryonuza ve dağıtım senaryolarınıza uygunsa, NATS sunucusunu doğrudan uygulamanızın içine gömebilirsiniz.

Embedding NATS, NATS docs

  • Ayrıca NATS'a özgü bir özellik vardır: Go ile yapılandırılmış uygulamalar için embedded mode desteklenmektedir.
  • Yani, mesaj brokerlarının genel yaklaşımı olan ayrı bir broker sunucusunun çalıştırılması ve uygulamanın istemcisi aracılığıyla bu sunucuyla iletişim kurulması yapısının aksine, broker'ın kendisi Go ile geliştirilmiş uygulamaya gömülebilir (embed) anlamına gelir.

Gömülü NATS'ın faydaları ve kullanım alanları

  • Konuyu iyi anlatan bir Youtube videosu bulunduğundan, açıklamayı videonun bağlantısıyla yetinerek geçiyorum.
  • Ayrı bir mesaj brokeri sunucusu dağıtmasanız bile modular monolith applictaion geliştirerek separate of concern hedefine ulaşabilir ve NATS'ı gömülü olarak uygulamanıza yerleştirmenin avantajlarından yararlanabilirsiniz. Buna ek olarak single binary deployment da mümkün hale gelir.
  • platform with no network (wasm) ortamlarının yanı sıra offline-first application uygulamalarında da yararlı biçimde kullanılabilir.

Resmi dokümandaki örnek

 1package main
 2
 3import (
 4    "fmt"
 5    "time"
 6
 7    "github.com/nats-io/nats-server/v2/server"
 8    "github.com/nats-io/nats.go"
 9)
10
11func main() {
12    opts := &server.Options{}
13
14    // Seçeneklerle yeni sunucuyu başlat
15    ns, err := server.NewServer(opts)
16
17    if err != nil {
18        panic(err)
19    }
20
21    // Sunucuyu goroutine aracılığıyla başlat
22    go ns.Start()
23
24    // Sunucunun bağlantılar için hazır olmasını bekle
25    if !ns.ReadyForConnections(4 * time.Second) {
26        panic("not ready for connection")
27    }
28
29    // Sunucuya bağlan
30    nc, err := nats.Connect(ns.ClientURL())
31
32    if err != nil {
33        panic(err)
34    }
35
36    subject := "my-subject"
37
38    // Konuya abone ol
39    nc.Subscribe(subject, func(msg *nats.Msg) {
40        // Mesaj verisini yazdır
41        data := string(msg.Data)
42        fmt.Println(data)
43
44        // Sunucuyu kapat (isteğe bağlı)
45        ns.Shutdown()
46    })
47
48    // Konuya veri yayınla
49    nc.Publish(subject, []byte("Hello embedded NATS!"))
50
51    // Sunucunun kapanmasını bekle
52    ns.WaitForShutdown()
53}
  • Bu, NATS resmi dokümanında yer alan Embedded NATS örneği olmakla birlikte, bu örnek koduyla ilerlendiğinde iletişim embedding mode ile gerçekleşmez.
1Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:34:20
2                                                                     in 0.017s (0)
3...
4tcp4       0      0  127.0.0.1.4222         127.0.0.1.63769        TIME_WAIT
  • watch 'netstat -an | grep 127.0.0.1' komutuyla localhost (127.0.0.1) üzerinden giden ağ trafiğini izlerken go run . ile ilgili go dosyası çalıştırıldığında, NATS'ın varsayılan portu olan 4222 üzerinden başlayan yeni ağ isteklerinin eklendiğini görebilirsiniz.

Embedding modu için doğru yapılandırmalar

  • Amaçlandığı gibi embedded mode ile iletişim kurmak için aşağıdaki iki seçeneğe ihtiyaç vardır:

    • Client: InProcessServer seçeneği eklenmelidir.
    • Server: Server.Options içinde DontListen adlı bayrak (flag) true olarak belirtilmelidir.

  • Bu kısımlar resmi olarak dokümante edilmemiştir; bu özelliğin başlangıcı ise şu PR üzerinden takip edilebilir.

    Bu PR üç şey ekliyor:

    1. Server üzerinde, TCP soketleri kullanmadan NATS sunucusuna bağlantı sağlayan bir net.Pipe oluşturan InProcessConn() fonksiyonu
    2. NATS sunucusuna olağan TCP dinleyicisinde dinleme yapmamasını söyleyen DontListen seçeneği
    3. AcceptLoop başlatılmadan hemen önce kapatılan ve readyForConnections fonksiyonunun beklediği startupComplete kanalı

    Bunun temel motivasyonu, tek süreçli (monolith) veya çok süreçli (polylith) modda çalışabilen bir uygulamamız olmasıdır. Basitlik için her iki mod için de NATS kullanmak istiyoruz; ancak monolith modunun, soket bağlantısı açmanın ya anlamsız olduğu (mobil) ya da tamamen mümkün olmadığı (WASM) çeşitli platformları desteklemesi gerekiyor. Bu değişiklikler, NATS'ı tamamen süreç içinde kullanmamıza olanak tanıyacak.

    İlgili nats-io/nats.go#774 PR'ı, istemci tarafına destek ekliyor.

    Bu projeye yaptığım ilk PR olduğundan, herhangi bir yerde bariz bir şeyi kaçırmışsam şimdiden özür dilerim.

    /cc @nats-io/core

Embedded mode için çalışan örnek

 1package main
 2
 3import (
 4	"fmt"
 5	"time"
 6
 7	"github.com/nats-io/nats-server/v2/server"
 8	"github.com/nats-io/nats.go"
 9)
10
11func main() {
12	opts := &server.Options{
13		// Gömülü NATS sunucusunu yapılandırmak için
14		// DonListen değerini true olarak ayarla
15		DontListen: true,
16	}
17
18	// Seçeneklerle yeni sunucuyu başlat
19	ns, err := server.NewServer(opts)
20
21	if err != nil {
22		panic(err)
23	}
24
25	// Sunucuyu goroutine aracılığıyla başlat
26	go ns.Start()
27
28	// Sunucunun bağlantılar için hazır olmasını bekle
29	if !ns.ReadyForConnections(10 * time.Second) {
30		panic("not ready for connection")
31	}
32
33	// Sunucuya süreç içi (in-process) bağlantı ile bağlan
34	nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
35
36	if err != nil {
37		panic(err)
38	}
39
40	subject := "my-subject"
41
42	// Konuya abone ol
43	nc.Subscribe(subject, func(msg *nats.Msg) {
44		// Mesaj verisini yazdır
45		data := string(msg.Data)
46		fmt.Println(data)
47
48		// Sunucuyu kapat (isteğe bağlı)
49		ns.Shutdown()
50	})
51
52	// Konuya veri yayınla
53	nc.Publish(subject, []byte("Hello embedded NATS!"))
54
55	// Sunucunun kapanmasını bekle
56	ns.WaitForShutdown()
57}
58

1Every 2.0s: netstat -an | grep 127.0.0.1         pravdalaptop-home.local: 02:37:50
2                                                                     in 0.023s (0)
3...no additional logs
4
  • Artık amaçlandığı gibi ek bir network hop oluşmadığını görebilirsiniz.

Perde arkası

TL;DR

diagram1

  • Bu kod main.go içinde çalıştırıldığında, dahili olarak hangi fonksiyonların nasıl çalıştığını gösteren sequence diagram'dır ve özü aşağıdaki gibidir.
    • DontListen: true ile sunucu, AcceptLoop adlı client listening phase aşamasını atlar.
    • Client'ın Connect seçeneklerinden InProcessServer etkinleştirilirse, bellek içi (in-memory) bir bağlantı oluşturulur, net.Pipe ile bir pipe yaratılır ve pipe'ın diğer ucu net.Conn tipinde client'a döndürülür.
    • Client ve server bu bağlantı üzerinden in-process iletişim gerçekleştirir.

Server

AcceptLoop

1// nats-server/server/server.go
2
3// Wait for clients.
4if !opts.DontListen {
5	s.AcceptLoop(clientListenReady)
6}
  • Öncelikle DontListen true ise, AcceptLoop adlı client listening phase aşaması atlanır.
 1// nats-server/server/server.go
 2
 3// AcceptLoop is exported for easier testing.
 4func (s *Server) AcceptLoop(clr chan struct{}) {
 5	// If we were to exit before the listener is setup properly,
 6	// make sure we close the channel.
 7	defer func() {
 8		if clr != nil {
 9			close(clr)
10		}
11	}()
12
13	if s.isShuttingDown() {
14		return
15	}
16
17	// Snapshot server options.
18	opts := s.getOpts()
19
20	// Setup state that can enable shutdown
21	s.mu.Lock()
22	hp := net.JoinHostPort(opts.Host, strconv.Itoa(opts.Port))
23	l, e := natsListen("tcp", hp)
24	s.listenerErr = e
25	if e != nil {
26		s.mu.Unlock()
27		s.Fatalf("Error listening on port: %s, %q", hp, e)
28		return
29	}
30	s.Noticef("Listening for client connections on %s",
31		net.JoinHostPort(opts.Host, strconv.Itoa(l.Addr().(*net.TCPAddr).Port)))
32
33	// Alert of TLS enabled.
34	if opts.TLSConfig != nil {
35		s.Noticef("TLS required for client connections")
36		if opts.TLSHandshakeFirst && opts.TLSHandshakeFirstFallback == 0 {
37			s.Warnf("Clients that are not using \"TLS Handshake First\" option will fail to connect")
38		}
39	}
40
41	// If server was started with RANDOM_PORT (-1), opts.Port would be equal
42	// to 0 at the beginning this function. So we need to get the actual port
43	if opts.Port == 0 {
44		// Write resolved port back to options.
45		opts.Port = l.Addr().(*net.TCPAddr).Port
46	}
47
48	// Now that port has been set (if it was set to RANDOM), set the
49	// server's info Host/Port with either values from Options or
50	// ClientAdvertise.
51	if err := s.setInfoHostPort(); err != nil {
52		s.Fatalf("Error setting server INFO with ClientAdvertise value of %s, err=%v", opts.ClientAdvertise, err)
53		l.Close()
54		s.mu.Unlock()
55		return
56	}
57	// Keep track of client connect URLs. We may need them later.
58	s.clientConnectURLs = s.getClientConnectURLs()
59	s.listener = l
60
61	go s.acceptConnections(l, "Client", func(conn net.Conn) { s.createClient(conn) },
62		func(_ error) bool {
63			if s.isLameDuckMode() {
64				// Signal that we are not accepting new clients
65				s.ldmCh <- true
66				// Now wait for the Shutdown...
67				<-s.quitCh
68				return true
69			}
70			return false
71		})
72	s.mu.Unlock()
73
74	// Let the caller know that we are ready
75	close(clr)
76	clr = nil
77}
  • Referans olarak, AcceptLoop fonksiyonu şu adımları gerçekleştirir. TLS veya hostPort gibi network iletişimiyle ilgili kısımlar, süreç içi iletişim yapıldığında gerekli olmadığından atlanabilir.

Client

InProcessServer

 1
 2// nats-go/nats.go
 3
 4// Connect will attempt to connect to the NATS system.
 5// The url can contain username/password semantics. e.g. nats://derek:pass@localhost:4222
 6// Comma separated arrays are also supported, e.g. urlA, urlB.
 7// Options start with the defaults but can be overridden.
 8// To connect to a NATS Server's websocket port, use the `ws` or `wss` scheme, such as
 9// `ws://localhost:8080`. Note that websocket schemes cannot be mixed with others (nats/tls).
10func Connect(url string, options ...Option) (*Conn, error) {
11	opts := GetDefaultOptions()
12	opts.Servers = processUrlString(url)
13	for _, opt := range options {
14		if opt != nil {
15			if err := opt(&opts); err != nil {
16				return nil, err
17			}
18		}
19	}
20	return opts.Connect()
21}

 1// nats-go/nats.go
 2
 3// Options can be used to create a customized connection.
 4type Options struct {
 5	// Url represents a single NATS server url to which the client
 6	// will be connecting. If the Servers option is also set, it
 7	// then becomes the first server in the Servers array.
 8	Url string
 9
10	// InProcessServer represents a NATS server running within the
11	// same process. If this is set then we will attempt to connect
12	// to the server directly rather than using external TCP conns.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • NATS server ile NATS client arasındaki bağlantıyı kuran Connect fonksiyonu, client URL'si ve connect Option'larını alır; bu Option'ları bir araya getiren Options struct'ında ise InProcesConnProvider interface tipinde InProcessServer adlı bir field bulunur.
1// main.go of example code
2
3// Initialize new server with options
4ns, err := server.NewServer(opts)
5
6//...
7
8// Connect to server via in-process connection
9nc, err := nats.Connect(ns.ClientURL(), nats.InProcessServer(ns))
  • NATS client tarafında Connect çağrılırken, InProcessServer field'ına nats.InProcessServer(ns) verildiğinde
 1// nats-go/nats.go
 2
 3// InProcessServer is an Option that will try to establish a direction to a NATS server
 4// running within the process instead of dialing via TCP.
 5func InProcessServer(server InProcessConnProvider) Option {
 6	return func(o *Options) error {
 7		o.InProcessServer = server
 8		return nil
 9	}
10}
  • Option'daki InProcessServer, gömülü NATS sunucusuyla değiştirilir ve
 1// nats-go/nats.go
 2
 3// createConn will connect to the server and wrap the appropriate
 4// bufio structures. It will do the right thing when an existing
 5// connection is in place.
 6func (nc *Conn) createConn() (err error) {
 7	if nc.Opts.Timeout < 0 {
 8		return ErrBadTimeout
 9	}
10	if _, cur := nc.currentServer(); cur == nil {
11		return ErrNoServers
12	}
13
14	// If we have a reference to an in-process server then establish a
15	// connection using that.
16	if nc.Opts.InProcessServer != nil {
17		conn, err := nc.Opts.InProcessServer.InProcessConn()
18		if err != nil {
19			return fmt.Errorf("failed to get in-process connection: %w", err)
20		}
21		nc.conn = conn
22		nc.bindToNewConn()
23		return nil
24	}
25
26	//...
27}
  • Bu interface, bağlantıyı oluşturan createConn fonksiyonunda, InProcessServer option'ı nil olmadığında (geçerli olduğunda) option içindeki InProcessServer'ın InProcesConn fonksiyonunu çalıştırarak
 1// nats-server/server/server.go
 2
 3// InProcessConn returns an in-process connection to the server,
 4// avoiding the need to use a TCP listener for local connectivity
 5// within the same process. This can be used regardless of the
 6// state of the DontListen option.
 7func (s *Server) InProcessConn() (net.Conn, error) {
 8	pl, pr := net.Pipe()
 9	if !s.startGoRoutine(func() {
10		s.createClientInProcess(pl)
11		s.grWG.Done()
12	}) {
13		pl.Close()
14		pr.Close()
15		return nil, fmt.Errorf("failed to create connection")
16	}
17	return pr, nil
18}
  • server'da uygulanan InProcessConn fonksiyonunu çağırır.
  • Bu fonksiyon, nats'ın Go client'ı olan nats.go içinde nc (nats connection) üzerindeki InProcessServer nil değilse çağrılır, connection (net.Conn) oluşturur ve bunu server'ın connection'ına bind eder.

Go'da tüketici tanımlı arayüz (consumer driven interface)

Bir tip, bir arayüzü yalnızca metodlarını uygulayarak gerçekleştirir. Açık bir niyet bildirimi, "implements" anahtar kelimesi yoktur. Örtük arayüzler, arayüzün tanımını uygulamasından ayırır; böylece bu arayüz, önceden bir düzenleme yapılmadan herhangi bir pakette yer alabilir.

Interfaces are implemented implicitly, A Tour of Go

Bir tip yalnızca bir arayüzü uygulamak için var oluyorsa ve bu arayüzün ötesinde hiçbir exported metodu olmayacaksa, tipin kendisini dışa aktarmaya gerek yoktur.

Generality, Effective Go

  • Bu interface tasarımı, Go'da sıkça bahsedilen consumer defined interface ve structural typing (duck typing) kavramlarını iyi yansıttığından, bu konuyu da birlikte tanıtmak istiyorum.
 1// nats-go/nats.go
 2
 3// Options can be used to create a customized connection.
 4type Options struct {
 5	// Url represents a single NATS server url to which the client
 6	// will be connecting. If the Servers option is also set, it
 7	// then becomes the first server in the Servers array.
 8	Url string
 9
10	// InProcessServer represents a NATS server running within the
11	// same process. If this is set then we will attempt to connect
12	// to the server directly rather than using external TCP conns.
13	InProcessServer InProcessConnProvider
14
15	//...
16}

1// nats-go/nats.go
2
3type InProcessConnProvider interface {
4	InProcessConn() (net.Conn, error)
5}
  • Koda geri dönelim. nats.go client'ında InProcessServer option struct field'ı, yalnızca InProcessConn işlevini gerçekleştiren InProcessConnProvider interface'i olarak tanımlanmıştır.
 1// nats-server/server/server.go
 2
 3// InProcessConn returns an in-process connection to the server,
 4// avoiding the need to use a TCP listener for local connectivity
 5// within the same process. This can be used regardless of the
 6// state of the DontListen option.
 7func (s *Server) InProcessConn() (net.Conn, error) {
 8	pl, pr := net.Pipe()
 9	if !s.startGoRoutine(func() {
10		s.createClientInProcess(pl)
11		s.grWG.Done()
12	}) {
13		pl.Close()
14		pr.Close()
15		return nil, fmt.Errorf("failed to create connection")
16	}
17	return pr, nil
18}
  • Ancak buna atanan tip, nats-server'daki Server olup, InProcessConn dışında birçok başka işlevi de yerine getirmektedir.
  • Çünkü bu durumda client'ın ilgilendiği tek şey InProcessConn adlı bir interface'in sağlanıp sağlanmadığıdır; diğerleri pek önemli değildir.
  • Bu nedenle nats.go client'ı, yalnızca InProcessConn() (net.Conn, error) işlevini tanımlayan InProcessConnProvider adlı consumer defined interface'i oluşturup kullanmaktadır.

Sonuç

  • NATS'ın embedded mode'unu, çalışma biçimini ve NATS kodu üzerinden incelenebilen Go'nun consumer defined interface yapısını kısaca ele aldık.
  • Bu bilgilerin, NATS'ı yukarıdaki amaçlarla kullanan kişilere yardımcı olmasını umuyor, yazıyı burada bitirmek istiyorum.