在Go语言中,使用net/rpc
包从服务器向客户端进行RPC调用是否真的可行?如果不行,有更好的解决方案吗?
我目前正在使用thrift (thrift4go) 用于服务器-客户端和客户端-服务器的RPC功能。默认情况下,thrift只执行类似于net/rpc的客户端-服务器调用。由于我还需要服务器-客户端通信,因此我进行了一些研究,并找到了bidi-thrift。Bidi-thrift解释了如何连接Java服务器+ Java客户端以实现双向thrift通信。
TCP连接具有一个传入和一个传出通信线路(RC和TX)。 bidi-thrift的思路是将RS和TX分开,并将其提供给客户端应用程序和服务器应用程序上的服务器(处理器)和客户端(远程)。我发现这在Go中很难做到。而且,这种方式没有“响应”可用(响应行正在使用中)。因此,服务中的所有方法都必须是“单向void”(即点火和忘记、调用不返回结果)。
我改变了bidi-thrift的想法,并使客户端打开两个与服务器的连接:A和B。第一个连接(A)用于执行客户端->服务器通信(客户端像往常一样进行调用)。第二个连接(B)被“劫持”,并连接到客户端上的服务器(处理器),同时连接到服务器上的客户端(远程)。我已经在Go服务器和Java客户端上实现了这种方式。它工作得非常好。速度快而可靠(就像普通thrift一样)。
一些资源...B连接(服务器-客户端)设置如下:
// factories
framedTransportFactory := thrift.NewTFramedTransportFactory(thrift.NewTTransportFactory())
protocolFactory := thrift.NewTBinaryProtocolFactoryDefault()
// create socket listener
addr, err := net.ResolveTCPAddr("tcp", "127.0.0.1:9091")
if err != nil {
log.Print("Error resolving address: ", err.Error(), "\n")
return
}
serverTransport, err := thrift.NewTServerSocketAddr(addr)
if err != nil {
log.Print("Error creating server socket: ", err.Error(), "\n")
return
}
// Start the server to listen for connections
log.Print("Starting the server for B communication (server->client) on ", addr, "\n")
err = serverTransport.Listen()
if err != nil {
log.Print("Error during B server: ", err.Error(), "\n")
return //err
}
// Accept new connections and handle those
for {
transport, err := serverTransport.Accept()
if err != nil {
return //err
}
if transport != nil {
// Each transport is handled in a goroutine so the server is availiable again.
go func() {
useTransport := framedTransportFactory.GetTransport(transport)
client := worldclient.NewWorldClientClientFactory(useTransport, protocolFactory)
// Thats it!
// Lets do something with the connction
result, err := client.Hello()
if err != nil {
log.Printf("Errror when calling Hello on client: %s\n", err)
}
// client.CallSomething()
}()
}
}
// preparations for B connection
TTransportFactory transportFactory = new TTransportFactory();
TProtocolFactory protocolFactory = new TBinaryProtocol.Factory();
YourServiceProcessor processor = new YourService.Processor<YourServiceProcessor>(new YourServiceProcessor(this));
/* Create thrift connection for B calls (server -> client) */
try {
// create the transport
final TTransport transport = new TSocket("127.0.0.1", 9091);
// open the transport
transport.open();
// add framing to the transport layer
final TTransport framedTransport = new TFramedTransport(transportFactory.getTransport(transport));
// connect framed transports to protocols
final TProtocol protocol = protocolFactory.getProtocol(framedTransport);
// let the processor handle the requests in new Thread
new Thread() {
public void run() {
try {
while (processor.process(protocol, protocol)) {}
} catch (TException e) {
e.printStackTrace();
} catch (NullPointerException e) {
e.printStackTrace();
}
}
}.start();
} catch(Exception e) {
e.printStackTrace();
}
Server.go
// server.go
package main
import (
"net"
"github.com/cenkalti/rpc2"
"fmt"
)
type Args struct{ A, B int }
type Reply int
func main(){
srv := rpc2.NewServer()
srv.Handle("add", func(client *rpc2.Client, args *Args, reply *Reply) error{
// Reversed call (server to client)
var rep Reply
client.Call("mult", Args{2, 3}, &rep)
fmt.Println("mult result:", rep)
*reply = Reply(args.A + args.B)
return nil
})
lis, _ := net.Listen("tcp", "127.0.0.1:5000")
srv.Accept(lis)
}
Client.go
// client.go
package main
import (
"fmt"
"github.com/cenkalti/rpc2"
"net"
)
type Args struct{ A, B int }
type Reply int
func main(){
conn, _ := net.Dial("tcp", "127.0.0.1:5000")
clt := rpc2.NewClient(conn)
clt.Handle("mult", func(client *rpc2.Client, args *Args, reply *Reply) error {
*reply = Reply(args.A * args.B)
return nil
})
go clt.Run()
var rep Reply
clt.Call("add", Args{5, 2}, &rep)
fmt.Println("add result:", rep)
}
RPC是一种(远程)服务。每当某台计算机请求远程服务时,它就充当客户端,请求服务器提供服务。在这个“定义”中,服务器调用客户端的RPC概念没有明确定义的含义。