Go语言RPC中函数序列化与GobEncoder的深度解析及替代方案

package common

// TaskRequest 定义了客户端发送的任务请求
type TaskRequest struct {
    FunctionName string // 要执行的函数名称
    Data         []byte // 函数所需的输入数据，可以是JSON、Gob或其他序列化格式
}

// TaskResponse 定义了工作节点返回的任务响应
type TaskResponse struct {
    Result []byte // 函数执行结果数据
    Error  string // 如果发生错误，则包含错误信息
}

package worker

import (
    "encoding/gob"
    "fmt"
    "log"
    "net"
    "net/rpc"
    "bytes"
    "your_project/common" // 假设 common 包在你的项目路径下
)

// WorkerService 是工作节点提供的RPC服务
type WorkerService struct {
    // functions 存储了所有可供远程调用的函数
    // 键是函数名称，值是实际的函数逻辑
    functions map[string]func(input []byte) ([]byte, error)
}

// NewWorkerService 创建并初始化 WorkerService
func NewWorkerService() *WorkerService {
    return &WorkerService{
        functions: make(map[string]func(input []byte) ([]byte, error)),
    }
}

// RegisterFunction 允许工作节点注册可执行的函数
func (ws *WorkerService) RegisterFunction(name string, fn func(input []byte) ([]byte, error)) {
    ws.functions[name] = fn
    log.Printf("Registered function: %s", name)
}

// ExecuteTask 是RPC方法，用于接收并执行任务
func (ws *WorkerService) ExecuteTask(req common.TaskRequest, resp *common.TaskResponse) error {
    log.Printf("Received task: FunctionName=%s, DataSize=%d", req.FunctionName, len(req.Data))

    fn, ok := ws.functions[req.FunctionName]
    if !ok {
        resp.Error = fmt.Sprintf("Function '%s' not found on worker.", req.FunctionName)
        return fmt.Errorf(resp.Error)
    }

    // 执行函数
    result, err := fn(req.Data)
    if err != nil {
        resp.Error = err.Error()
        return err
    }

    resp.Result = result
    return nil
}

// StartRPCServer 启动RPC服务器
func StartRPCServer(addr string, ws *WorkerService) {
    rpc.Register(ws)
    listener, err := net.Listen("tcp", addr)
    if err != nil {
        log.Fatalf("Failed to listen: %v", err)
    }
    log.Printf("Worker RPC server listening on %s", addr)
    for {
        conn, err := listener.Accept()
        if err != nil {
            log.Printf("Accept error: %v", err)
            continue
        }
        go rpc.ServeConn(conn)
    }
}

// 示例：一个具体的函数实现，用于计算两个整数的和
func sumNumbers(input []byte) ([]byte, error) {
    var nums struct {
        A int
        B int
    }
    decoder := gob.NewDecoder(bytes.NewReader(input))
    if err := decoder.Decode(&nums); err != nil {
        return nil, fmt.Errorf("failed to decode input for sumNumbers: %v", err)
    }

    sum := nums.A + nums.B
    log.Printf("Executing sumNumbers: %d + %d = %d", nums.A, nums.B, sum)

    var buf bytes.Buffer
    encoder := gob.NewEncoder(&buf)
    if err := encoder.Encode(sum); err != nil {
        return nil, fmt.Errorf("failed to encode result for sumNumbers: %v", err)
    }
    return buf.Bytes(), nil
}

// 示例：工作节点主函数
func main() {
    ws := NewWorkerService()
    ws.RegisterFunction("sumNumbers", sumNumbers) // 注册一个函数

    // 可以在这里注册更多函数...

    StartRPCServer(":1234", ws)
}

package client

import (
    "encoding/gob"
    "fmt"
    "log"
    "net/rpc"
    "bytes"
    "your_project/common" // 假设 common 包在你的项目路径下
)

func main() {
    client, err := rpc.Dial("tcp", "localhost:1234")
    if err != nil {
        log.Fatalf("Failed to dial RPC server: %v", err)
    }
    defer client.Close()

    // 准备输入数据
    var inputData struct {
        A int
        B int
    }
    inputData.A = 10
    inputData.B = 20

    var buf bytes.Buffer
    encoder := gob.NewEncoder(&buf)
    if err := encoder.Encode(inputData); err != nil {
        log.Fatalf("Failed to encode input data: %v", err)
    }

    // 构建任务请求
    req := common.TaskRequest{
        FunctionName: "sumNumbers",
        Data:         buf.Bytes(),
    }

    var resp common.TaskResponse
    err = client.Call("WorkerService.ExecuteTask", req, &resp)
    if err != nil {
        log.Fatalf("RPC call failed: %v", err)
    }

    if resp.Error != "" {
        log.Fatalf("Worker reported an error: %s", resp.Error)
    }

    // 解析结果
    var sumResult int
    decoder := gob.NewDecoder(bytes.NewReader(resp.Result))
    if err := decoder.Decode(&sumResult); err != nil {
        log.Fatalf("Failed to decode result: %v", err)
    }

    fmt.Printf("Task 'sumNumbers' executed successfully. Result: %d\n", sumResult)

    // 尝试调用一个不存在的函数
    req2 := common.TaskRequest{
        FunctionName: "nonExistentFunction",
        Data:         []byte{},
    }
    var resp2 common.TaskResponse
    err = client.Call("WorkerService.ExecuteTask", req2, &resp2)
    if err != nil {
        fmt.Printf("Expected RPC call failure for non-existent function: %v\n", err)
    }
    if resp2.Error != "" {
        fmt.Printf("Worker reported error for non-existent function: %s\n", resp2.Error)
    }
}