墨 痕

func (t *T) MethodName(argType T1, replyType *T2) error

package common

// RPC请求参数
type AgentReportRequest struct {
    Hostname string `json:"hostname"`
    IP       string `json:"ip"`
    Version  string `json:"version"`
}

// RPC响应参数
// code == 0 -> success
// code == 1 -> fail
type RpcResponse struct {
    Code int    `json:"code"`
    Msg  string `json:"msg"`
}

package server

import (
    "github.com/mogl/jsonrpc/common"
    "log"
)

type Agent int

// 方法需要满足之前提到的5个条件
// 从client传入的args中 获取请求参数，进行逻辑处理，并构造reply结构返回
func (ag *Agent) ReportStatus(args *common.AgentReportRequest, reply *common.RpcResponse) error {
    if args.Hostname == "" {
        reply.Code = 1
        return nil
    }

    log.Printf("rpc server receive, hostname:%s, ip:%s, version:%s", args.Hostname, args.IP, args.Version)
    reply.Code = 0
    reply.Msg = "call rpc server successfully."

    return nil
}

package server

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

// 创建rpc server服务
func Start() {
    addr := ":2333"

    // 创建server
    server := rpc.NewServer()

    // 注册处理器
    server.Register(new(Agent)) // server/agent.go

    // 监听TCP端口
    l, e := net.Listen("tcp", addr)
    if e != nil {
        log.Fatalln("listen error:", e)
    } else {
        log.Println("listerning", addr)
    }

    // 等待请求、每个client请求新建goroutine处理
    for {
        conn, err := l.Accept()
        if err != nil {
            log.Fatalln("listener accept fail:", err)
            time.Sleep(time.Duration(100) * time.Millisecond)
            continue
        }
        // 绑定rpc的编码器
        // 新建jsonrpc编码器，并将该编码器绑定给http处理器
        go server.ServeCodec(jsonrpc.NewServerCodec(conn))
    }
}

package client

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

// 自定义client连接用的结构体
type ConnRpcClient struct {
    sync.RWMutex
    rpcClient *rpc.Client
    RpcServer string
    Timeout   time.Duration
}

// client和server建立连接
func (c *ConnRpcClient) rpcServerConn() error {
    if c.rpcClient != nil {
        return nil
    }

    // 重试次数
    retry := 1

    for {
        if c.rpcClient != nil {
            return nil
        }

        // TCP连接
        conn, err := net.DialTimeout("tcp", c.RpcServer, c.Timeout)
        if err != nil {
            log.Printf("dial %s fail: %v", c.RpcServer, err)
            // 最多尝试建立连接3次
            if retry > 3 {
                log.Printf("dial %s fail(retry %d times): %v", c.RpcServer, retry, err)
                return err
            }
            time.Sleep(time.Second << uint(retry)) // 指数退避重试
            retry++
            continue
        } else {
            c.rpcClient = jsonrpc.NewClient(conn) // 基于TCP连接建立rpcClient实例
        }
        return err
    }
}

// client调用server方法
func (c *ConnRpcClient) Call(method string, args interface{}, reply interface{}) error {
    // 一个client实例如果要串行调用，同步阻塞，则加锁
    //c.Lock()
    //defer c.Unlock()

    // client和server建立连接
    err := c.rpcServerConn()
    if err != nil {
        log.Fatalf("connect rpc server %s fail: %v", c.RpcServer, err)
        return err
    }

    // 远程调用超时
    timeout := time.Duration(15) * time.Second
    done := make(chan error, 1)

    // client远程调用方法
    go func() {
        // 同步方式调用
        err := c.rpcClient.Call(method, args, reply)
        done <- err
    }()

    // 调用超时机制
    select {
    case <-time.After(timeout):
        log.Printf("rpc call timeout: %s -> %s", c.rpcClient, c.RpcServer)
        c.rpcClient.Close()
        return fmt.Errorf("rpc call timeout: %s -> %s", c.rpcClient, c.RpcServer)
    case err := <-done:
        if err != nil {
            c.rpcClient.Close()
            return err
        }
    }

    return nil
}

package main

import (
    "github.com/mogl/jsonrpc/server"
)

func main() {
    go server.Start()   // 创建jsonrpc server服务

    select {}
}

package main

import (
    "github.com/mogl/jsonrpc/client"
    "github.com/mogl/jsonrpc/common"
    "log"
    "time"
)

func main() {
    // 请求参数
    req := &common.AgentReportRequest{
        Hostname: "sai",
        IP:       "localhost",
        Version:  "1.0",
    }

    // 响应参数
    resp := &common.RpcResponse{}

    // client实例
    rpcClient := client.ConnRpcClient{
        RpcServer: ":2333",
        Timeout:   time.Duration(10) * time.Second,
    }

    time.Sleep(time.Duration(3 * time.Second))
    rpcClient.Call("Agent.ReportStatus", req, resp)     // jsonrpc 调用

    log.Println("response code:", resp.Code)
    log.Println("response msg:", resp.Msg)
}