一文吃透Go的内置RPC原理

package common

type Args struct {
	A, B int
}

type Quotient struct {
	Quo, Rem int
}

type Arith struct{}

func (t *Arith) Multiply(args *common.Args, reply *int) error {
	*reply = args.A * args.B
	return nil
}

func (t *Arith) Divide(args *common.Args, quo *common.Quotient) error {
	if args.B == 0 {
		return errors.New("divide by zero")
	}
	quo.Quo = args.A / args.B
	quo.Rem = args.A % args.B
	return nil
}

func main() {
	arith := new(Arith)
	rpc.Register(arith)
	rpc.HandleHTTP()
	l, e := net.Listen("tcp", ":9876")
	if e != nil {
		panic(e)
	}

	go http.Serve(l, nil)

	var wg sync.WaitGroup
	wg.Add(1)
	wg.Wait()
}

func main() {
	client, err := rpc.DialHTTP("tcp", "127.0.0.1:9876")
	if err != nil {
		panic(err)
	}

	args := common.Args{A: 7, B: 8}
	var reply int
  // 同步调用
	err = client.Call("Arith.Multiply", &args, &reply)
	if err != nil {
		panic(err)
	}
	fmt.Printf("Call Arith: %d * %d = %d\n", args.A, args.B, reply)

  // 异步调用
	quotient := new(common.Quotient)
	divCall := client.Go("Arith.Divide", args, quotient, nil)
	replyCall := <-divCall.Done

	fmt.Printf("Go Divide: %d divide %d = %+v %+v\n", args.A, args.B, replyCall.Reply, quotient)
}

rpc.Register(arith)

rpc.HandleHTTP()

type Handler interface {
	ServeHTTP(ResponseWriter, *Request)
}

// ServeHTTP implements an http.Handler that answers RPC requests.
func (server *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	// ①
  if req.Method != "CONNECT" {
		w.Header().Set("Content-Type", "text/plain; charset=utf-8")
		w.WriteHeader(http.StatusMethodNotAllowed)
		io.WriteString(w, "405 must CONNECT\n")
		return
	}
  // ②
	conn, _, err := w.(http.Hijacker).Hijack()
	if err != nil {
		log.Print("rpc hijacking ", req.RemoteAddr, ": ", err.Error())
		return
	}
  // ③
	io.WriteString(conn, "HTTP/1.0 "+connected+"\n\n")
	// ④
	server.ServeConn(conn)
}

type Request struct {
	// Method specifies the HTTP method (GET, POST, PUT, etc.).
	// For client requests, an empty string means GET.
	//
	// Go's HTTP client does not support sending a request with
	// the CONNECT method. See the documentation on Transport for
	// details.
	Method string
}

"HTTP/1.0 200 Connected to Go RPC \n\n"

type Server struct {
	...
	freeReq    *Request
	..
}

type Request struct {
	ServiceMethod string 
	Seq           uint64
	next          *Request
}

func (server *Server) getRequest() *Request {
	server.reqLock.Lock()
	req := server.freeReq
	if req == nil {
		req = new(Request)
	} else {
		server.freeReq = req.next
		*req = Request{}
	}
	server.reqLock.Unlock()
	return req
}

func (server *Server) freeRequest(req *Request) {
	server.reqLock.Lock()
	req.next = server.freeReq
	server.freeReq = req
	server.reqLock.Unlock()
}

type ClientCodec interface {
	WriteRequest(*Request, interface{}) error
	ReadResponseHeader(*Response) error
	ReadResponseBody(interface{}) error

	Close() error
}

type ServerCodec interface {
	ReadRequestHeader(*Request) error
	ReadRequestBody(interface{}) error
	WriteResponse(*Response, interface{}) error
  
	Close() error
}

func (c *gobClientCodec) WriteRequest(r *Request, body interface{}) (err error) {
	if err = c.enc.Encode(r); err != nil {
		return
	}
	if err = c.enc.Encode(body); err != nil {
		return
	}
	return c.encBuf.Flush()
}

func (c *gobClientCodec) ReadResponseHeader(r *Response) error {
	return c.dec.Decode(r)
}

func (c *gobClientCodec) ReadResponseBody(body interface{}) error {
	return c.dec.Decode(body)
}