Remote Procedure Calls (RPC) in Go

Aus archium.org
Wechseln zu: Navigation, Suche


Bedeutung

Was ist dieses RPC eigentlich?

Remote Procedure Call bedeutet soviel wie "Aufruf einer fernen Prozedur." Die Implementierung einer solchen Prozedur ist in vielen Anwendungen nicht mehr wegzudenken. Zum Beispiel immer dann, wenn ein Informationsaustausch zwischen Clients vollkommen eigenständiger Prozesse, die in verschiedenen Programmen und sogar auf entfernten Rechnern laufen, stattfinden soll. Es wird häufig genutzt, um Client-Server-Modelle aufzubauen. RPC basiert auf UDP oder TCP. Daher kann man auch von einer verbindungslosen Kommunikation sprechen.

Das User Datagram Protocol (UDP) und das Transmission Control Protocol (TCP) gehören zur TCP/IP-Protokollfamilie und regeln den Umgang mit und die Anordnung von Datenpaketen bei der Kommunikation im Internet. TCP/IP ist eine Client-Server Kommunikation, sie basiert auf einer exklusiven Verbindung von Host zu Host. Damit ein Host mehrere Verbindungen gleichzeitig bedienen kann, werde den Services bestimmte Ports zugeordnet. Die Datenströme sind in zwei Schichten aufgeteilt:

  1. In der Datenschicht wird eine Nachricht vor dem Versand in kleinere Pakete zerlegt.
  2. In der IP-Schicht stehen die Adressinformationen.[1]
TCP
steuert den Datenfluß und verhindert den Paketverlust. TCP nimmt ein File bzw. den Datenstrom von den Anwendungen am Port entgegen, verteilt die Pakete und stellt ihnen einen IP-Header bei. Der Empfänger erhält mit dem Header eine Anleitung, um die Datenpakete wieder in der richtigen Reihenfolge und vollständig zusammenzusetzen.
UDP
hat die selbe Aufgabe wie TCP, aber es fehlen die Kontrollfunktionen und auch die Nummerierung der Pakete entfällt. Es ist dadurch schlanker, einfacher zu verarbeiten und deswegen auch schneller. Es eignet sich für Anwendungen, die nur einzelne Datenpakete ohne Zusammenhang transportieren (z.B. Zeitmarken) oder für den schnellen Transport großer Datenmengen mit Anwendungen, die die komplexen Kontrollfunktionen selber auf eigene Weise übernehmen (z.B. Network File System NFS).

Funktionsweise

Was macht so ein Remote Procedure Call?

Die Kommunikation beginnt, indem der Client eine Anfrage an einen bekannten Server schickt und auf die Antwort wartet. In der Anfrage gibt der Client an, welche Funktion mit welchen Parametern ausgeführt werden soll. Der Server bearbeitet die Anfrage und schickt die Antwort an den Client zurück. Nach Empfang der Nachricht kann der Client seine Verarbeitung fortführen.[2]

Beispiele für RPC mit Go

Nachfolgende Beispiele bestehen immer aus zwei voneinander unabhängig auszuführenden Programmen, dem Server und dem Client. Der Server ist immer vor dem Client zu starten.

Variante 1-1: Package "net/rpc"

[Server] Datei "main.go"

package main

import (
	"errors"
	"fmt"
	"log"
	"net"
	"net/rpc"
)

type QuotientServer struct {
	Dividend, Divisor int
}

type ServerMethods int

func (rcv *ServerMethods) Divide(args QuotientServer, result *float64) error {
	if args.Divisor == 0 {
		return errors.New("division by zero")
	} else {
		*result = float64(args.Dividend) / float64(args.Divisor)
		return nil
	}
}
func main() {
	rpc.Register(new(ServerMethods))

	l, e := net.Listen("tcp", "127.0.0.1:17152")

	defer l.Close()

	if e != nil {
		log.Fatal("listen error:", e)
	}

	for {
		c, err := l.Accept()
		fmt.Printf("request from %v\n", c.RemoteAddr())
		if err != nil {
			continue
		}
		go rpc.ServeConn(c)
	}
}
Ergebnis
$ ./rpc_server 
request from 127.0.0.1:47156
request from 127.0.0.1:47158
request from 127.0.0.1:47160
^C


[Client] Datei "main.go"

package main

import (
	"fmt"
	"log"
	"net/rpc"
	"time"
)

type quotientClient struct {
	Dividend, Divisor int
}

func main() {
	t0 := time.Now()
	var (
		result float64
		args   quotientClient
	)

	args.Dividend = 7
	args.Divisor = 8

	client, err := rpc.Dial("tcp", "127.0.0.1:17152")
	if err != nil {
		log.Fatal("dial error:", err)
	}

	err = client.Call("ServerMethods.Divide", &args, &result)
	//err = client.Call("ServerMethods.Divide", args, &result) // Valid, too
	if err != nil {
		log.Fatal("division error:", err)
	}
	fmt.Printf("request took %v\n", time.Since(t0))
	fmt.Printf("%d / %d = %0.3f\n", args.Dividend, args.Divisor, result)
}
Ergebnis
$ ./rpc_client 
request took 405.45µs
7 / 8 = 0.875
$ ./rpc_client 
request took 1.055826ms
7 / 8 = 0.875
$ ./rpc_client 
request took 1.089143ms
7 / 8 = 0.875
$

Variante 1-2: Package "net/rpc" mit HTTP-Handler

ServeConn aus dem oberen Beispiel liefert eine einzelne Verbindung, die nach Beendigung neu gestartet werden muß. Demgegenüber ermöglicht der HTTP-Handler den Aufbau eines dauerhaft "lauschenden" Servers. Er wird nur einmal gestartet und bedient beliebig viele Requests. Allerdings stehen die Methoden des Listeners nicht mehr zur Verfügung.

[Server] Datei "main.go"

package main

import (
	"errors"
	"log"
	"net"
	"net/http"
	"net/rpc"
)

type QuotientServer struct {
	Dividend, Divisor int
}

type ServerMethods int

func (rcv *ServerMethods) Divide(args QuotientServer, result *float64) error {
	if args.Divisor == 0 {
		return errors.New("division by zero")
	} else {
		*result = float64(args.Dividend) / float64(args.Divisor)
		return nil
	}
}
func main() {
	rpc.Register(new(ServerMethods))
	rpc.HandleHTTP()

	l, e := net.Listen("tcp", "127.0.0.1:17152")

	defer l.Close()

	if e != nil {
		log.Fatal("listen error:", e)
	}

	go http.Serve(l, nil)

	useChannelToWait := make(chan int, 1)
	<-useChannelToWait
}
Ergebnis
$ ./rpc_server 
^C

[Client] Datei "main.go"

package main

import (
	"fmt"
	"log"
	"net/rpc"
	"time"
)

type quotientClient struct {
	Dividend, Divisor int
}

func main() {
	t0 := time.Now()
	var (
		result float64
		args   quotientClient
	)

	args.Dividend = 7
	args.Divisor = 8

	client, err := rpc.DialHTTP("tcp", "127.0.0.1:17152")
	if err != nil {
		log.Fatal("dial error:", err)
	}

	err = client.Call("ServerMethods.Divide", &args, &result)
	//err = client.Call("ServerMethods.Divide", args, &result) // Valid, too
	if err != nil {
		log.Fatal("division error:", err)
	}
	fmt.Printf("request took %v\n", time.Since(t0))
	fmt.Printf("%d / %d = %0.3f\n", args.Dividend, args.Divisor, result)
}
Ergebnis
$ ./rpc_client
request took 1.311427ms
7 / 8 = 0.875
$ ./rpc_client
request took 576.049µs
7 / 8 = 0.875
$ ./rpc_client
request took 1.224976ms
7 / 8 = 0.875
$

Variante 1-3: Package "net/rpc/jsonrpc"

"net/rpc" kommuniziert nur von Go-Prozessen zu Go-Prozessen. Für cross-language- / cross-platform-Kommunikation entsprechend der JSON-RPC 2.0 Specification eignet sich stattdessen "net/rpc/jsonrpc".

[Server] Datei "main.go"

package main

import (
	"errors"
	"fmt"
	"log"
	"net"
	"net/rpc"
	"net/rpc/jsonrpc"
)

type QuotientServer struct {
	Dividend, Divisor int
}

type ServerMethods int

func (rcv *ServerMethods) Divide(args QuotientServer, result *float64) error {
	if args.Divisor == 0 {
		return errors.New("division by zero")
	} else {
		*result = float64(args.Dividend) / float64(args.Divisor)
		return nil
	}
}
func main() {
	rpc.Register(new(ServerMethods))

	l, e := net.Listen("tcp", "127.0.0.1:17152")

	defer l.Close()

	if e != nil {
		log.Fatal("listen error:", e)
	}

	for {
		c, err := l.Accept()
		fmt.Printf("request from %v\n", c.RemoteAddr())
		if err != nil {
			continue
		}
		go jsonrpc.ServeConn(c)
	}
}


[Client] Datei "main.go"

package main

import (
	"fmt"
	"log"
	"net/rpc/jsonrpc"
	"time"
)

type quotientClient struct {
	Dividend, Divisor int
}

func main() {
	t0 := time.Now()
	var (
		result float64
		args   quotientClient
	)

	args.Dividend = 7
	args.Divisor = 8

	client, err := jsonrpc.Dial("tcp", "127.0.0.1:17152")
	if err != nil {
		log.Fatal("dial error:", err)
	}

	err = client.Call("ServerMethods.Divide", &args, &result)
	//err = client.Call("ServerMethods.Divide", args, &result) // Valid, too
	if err != nil {
		log.Fatal("division error:", err)
	}
	fmt.Printf("request took %v\n", time.Since(t0))
	fmt.Printf("%d / %d = %0.3f\n", args.Dividend, args.Divisor, result)
}

Variante 1-4: Package "golang.org/x/net/websocket" in Kombination mit "net/rpc" oder "net/rpc/jsonrpc"

[Server] Datei "main.go"

package main

import (
	"errors"
	"fmt"
	"net/http"
	"net/rpc"

	"golang.org/x/net/websocket"
)

type QuotientServer struct {
	Dividend, Divisor int
}

type ServerMethods int

func (rcv *ServerMethods) Divide(args QuotientServer, result *float64) error {
	if args.Divisor == 0 {
		return errors.New("division by zero")
	} else {
		*result = float64(args.Dividend) / float64(args.Divisor)
		return nil
	}
}
func main() {
	rpc.Register(new(ServerMethods))

	http.Handle("/ws", websocket.Handler(func(connection *websocket.Conn) {
		fmt.Printf("request from %v\n", connection.RemoteAddr())
		rpc.ServeConn(connection)       // variant 1
		//jsonrpc.ServeConn(connection) // variant 2
	}))
	http.ListenAndServe("127.0.0.1:17152", nil)

}


[Client] Datei "main.go"

package main

import (
	"fmt"
	"log"
	"net/rpc"
	"time"

	"golang.org/x/net/websocket"
)

type quotientClient struct {
	Dividend, Divisor int
}

func main() {
	t0 := time.Now()
	var (
		result float64
		args   quotientClient
	)

	args.Dividend = 7
	args.Divisor = 8

	connection, err := websocket.Dial("ws://127.0.0.1:17152/ws", "", "http://127.0.0.1")
	if err != nil {
		log.Fatal("dial error:", err)
	}

	defer connection.Close()

	client := rpc.NewClient(connection)       // variant 1
	//client := jsonrpc.NewClient(connection) // variant 2

	err = client.Call("ServerMethods.Divide", &args, &result)
	//err = client.Call("ServerMethods.Divide", args, &result) // Valid, too

	if err != nil {
		log.Fatal("division error:", err)
	}
	fmt.Printf("request took %v\n", time.Since(t0))
	fmt.Printf("%d / %d = %0.3f\n", args.Dividend, args.Divisor, result)
}

Websockets als Alternative – geeignet zur Einbettung von WebAssembly in verteilte Systeme

Die Kommunikation zwischen Go-Programmen und Go-WebAssembly-Anwendungen über RPC ist nicht zulässig.[3]Mit Hilfe der minimalistischen Websocket-Library für Go von https://github.com/nhooyr/websocket können wir eine ähnlich komfortable, bidirektionale Kommunikation zwischen zwei Go-Programmen (auch WASM!) aufbauen.

Variante 2-1: Byte-Stream

[Server] Datei "main.go"

package main

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"nhooyr.io/websocket"
)

func main() {
	ReverseFunc := func(s string) (result string) {
		for _, v := range s {
			result = string(v) + result
		}
		return
	}

	hf := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

		var (
			buffer []byte
			mt     websocket.MessageType
		)

		connection, err := websocket.Accept(w, r, &websocket.AcceptOptions{
			CompressionMode: websocket.CompressionDisabled,
			OriginPatterns:  []string{"*"},
		})

		if err != nil {
			panic(err)
		}
		defer connection.Close(websocket.StatusInternalError, "Websocket: internal Error")

		ctx, cancel := context.WithTimeout(r.Context(), time.Second*10)
		defer cancel()

		//Receive
		mt, buffer, err = connection.Read(ctx)
		if err != nil {
			panic(err)
		}
		fmt.Printf("Message received: %s, message type %d\n", string(buffer), mt)

		//Send
		connection.Write(ctx, websocket.MessageText, []byte(ReverseFunc(string(buffer))))
		fmt.Printf("Message send: %s\n", ReverseFunc(string(buffer)))

		connection.Close(websocket.StatusNormalClosure, "")
	})

	http.ListenAndServe("127.0.0.1:17152", hf)
}

[Client] Datei "main.go" (Der WebAssembly-Partner muß der Client sein)

package main

import (
	"context"
	"fmt"
	"time"

	"nhooyr.io/websocket"
)

const MyMessage = "looc tsi muihcra"

func main() {
	var (
		buffer []byte
		mt     websocket.MessageType
	)

	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	connection, _, err := websocket.Dial(ctx, "ws://127.0.0.1:17152", nil)
	if err != nil {
		panic(err)
	}
	defer connection.Close(websocket.StatusInternalError, "Websocket: internal Error")

	//Send
	connection.Write(ctx, websocket.MessageBinary, []byte(MyMessage))
	fmt.Printf("Message send: %s\n", MyMessage)

	//Receive
	mt, buffer, err = connection.Read(ctx)
	if err != nil {
		panic(err)
	}
	fmt.Printf("Message received: %s, message type %d\n", string(buffer), mt)

	connection.Close(websocket.StatusNormalClosure, "")
}

Variante 2-2: JSON-Text

[Server] Datei "main.go"

package main

import (
	"context"
	"fmt"
	"net/http"
	"time"

	"nhooyr.io/websocket"
	"nhooyr.io/websocket/wsjson"
)

func main() {
	ReverseFunc := func(s string) (result string) {
		for _, v := range s {
			result = string(v) + result
		}
		return
	}

	hf := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

		var buffer interface{}

		connection, err := websocket.Accept(w, r, &websocket.AcceptOptions{
			CompressionMode: websocket.CompressionDisabled,
			OriginPatterns:  []string{"*"},
		})

		if err != nil {
			panic(err)
		}
		defer connection.Close(websocket.StatusInternalError, "Websocket: internal Error")

		ctx, cancel := context.WithTimeout(r.Context(), time.Second*10)
		defer cancel()

		//Receive
		err = wsjson.Read(ctx, connection, &buffer)
		fmt.Printf("Message received: %s\n", buffer.(string))

		//Send
		err = wsjson.Write(ctx, connection, ReverseFunc(string(buffer.(string))))
		if err != nil {
			panic(err)
		}
		fmt.Printf("Message send: %s\n", ReverseFunc(string(buffer.(string))))

		connection.Close(websocket.StatusNormalClosure, "")
	})

	http.ListenAndServe("127.0.0.1:17152", hf)
}

[Client] Datei "main.go" (Der WebAssembly-Partner muß der Client sein)

package main

import (
	"context"
	"fmt"
	"time"

	"nhooyr.io/websocket"
	"nhooyr.io/websocket/wsjson"
)

const MyMessage = "looc tsi muihcra"

func main() {
	var buffer interface{}

	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	connection, _, err := websocket.Dial(ctx, "ws://127.0.0.1:17152", nil)
	if err != nil {
		panic(err)
	}
	defer connection.Close(websocket.StatusInternalError, "Websocket: internal Error")

	//Send
	err = wsjson.Write(ctx, connection, MyMessage)
	if err != nil {
		panic(err)
	}
	fmt.Printf("Message send: %s\n", MyMessage)

	//Receive
	err = wsjson.Read(ctx, connection, &buffer)
	fmt.Printf("Message received: %s\n", buffer.(string))

	connection.Close(websocket.StatusNormalClosure, "")
}

Anmerkungen