使用 Go net/rpc 实现分布式消息通信与确认机制

package main

import (
    "log"
    "net"
    "net/http"
    "net/rpc"
    "time" // 引入time包用于模拟耗时操作
)

// Args 定义远程方法接收的参数结构体
type Args struct {
    A, B int
}

// Reply 定义远程方法返回的结果结构体
// 在本示例中，我们直接使用int作为reply，但复杂场景下建议使用结构体
// type Reply struct {
//     Result int
//     Status string
// }

// Arith 是一个示例服务，提供了算术运算
type Arith int

// Multiply 是 Arith 服务的一个方法，用于计算两个整数的乘积
func (t *Arith) Multiply(args *Args, reply *int) error {
    log.Printf("Server received Multiply call with A=%d, B=%d", args.A, args.B)
    time.Sleep(100 * time.Millisecond) // 模拟耗时操作
    *reply = args.A * args.B
    log.Printf("Server responded with result: %d", *reply)
    return nil
}

// Sum 是 Arith 服务的一个方法，用于计算两个整数的和
func (t *Arith) Sum(args *Args, reply *int) error {
    log.Printf("Server received Sum call with A=%d, B=%d", args.A, args.B)
    time.Sleep(50 * time.Millisecond) // 模拟耗时操作
    *reply = args.A + args.B
    log.Printf("Server responded with result: %d", *reply)
    return nil
}

func main() {
    // 1. 实例化服务
    arith := new(Arith)

    // 2. 注册服务
    // rpc.Register() 注册的服务名默认为结构体类型名，即 "Arith"
    err := rpc.Register(arith)
    if err != nil {
        log.Fatalf("Error registering RPC service: %v", err)
    }

    // 3. 配置并启动监听器
    // rpc.HandleHTTP() 将 RPC 服务暴露在 HTTP 路径 /_goRPC 上
    rpc.HandleHTTP()

    // 监听 TCP 端口
    listenPort := ":1234"
    l, err := net.Listen("tcp", listenPort)
    if err != nil {
        log.Fatalf("Listen error on port %s: %v", listenPort, err)
    }
    log.Printf("RPC server listening on %s", listenPort)

    // 4. 在新的 Goroutine 中启动 HTTP 服务器，处理 RPC 请求
    // http.Serve() 会阻塞，因此需要放在 Goroutine 中
    go http.Serve(l, nil)

    // 保持主 Goroutine 运行，等待服务中断信号（例如 Ctrl+C）
    select {}
}

package main

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

    // 引入server包，以便使用其定义的Args结构体
    // 实际项目中，Args结构体通常会放在一个共享的包中
    // 这里为了示例方便，假设server.Args是可访问的
    // 如果是独立项目，需要复制Args定义或使用go modules共享
    "your_module_path/server_example" // 替换为你的实际模块路径
)

// 假设server_example包中定义了Args结构体
// type Args struct {
//     A, B int
// }

func main() {
    serverAddress := "127.0.0.1" // RPC 服务器地址
    serverPort := "1234"

    // 1. 连接到 RPC 服务器
    // rpc.DialHTTP() 用于连接通过 HTTP 暴露的 RPC 服务
    client, err := rpc.DialHTTP("tcp", serverAddress+":"+serverPort)
    if err != nil {
        log.Fatalf("Error dialing RPC server at %s:%s: %v", serverAddress, serverPort, err)
    }
    defer client.Close() // 确保连接关闭

    log.Printf("Successfully connected to RPC server at %s:%s", serverAddress, serverPort)

    // 2. 发起同步远程调用
    callMultiply(client)
    callSum(client)

    // 3. 异步远程调用示例
    callAsyncMultiply(client)

    // 4. 发送消息到多个主机（模拟）
    // 假设有多个RPC服务器地址
    otherServerAddresses := []string{
        "127.0.0.1:1235", // 假设有另一个服务器运行在1235端口
        "127.0.0.1:1236", // 假设有第三个服务器运行在1236端口
    }
    sendMessageToMultipleHosts(otherServerAddresses)

    fmt.Println("\nAll RPC calls completed.")
}

// callMultiply 示例：同步调用 Multiply 方法
func callMultiply(client *rpc.Client) {
    args := &server_example.Args{A: 7, B: 8} // 使用server_example.Args
    var reply int // 接收返回结果的变量
    log.Printf("Client calling Arith.Multiply with A=%d, B=%d", args.A, args.B)
    err := client.Call("Arith.Multiply", args, &reply) // "Arith" 是服务名，"Multiply" 是方法名
    if err != nil {
        log.Fatalf("Error calling Arith.Multiply: %v", err)
    }
    fmt.Printf("Arith: %d * %d = %d\n", args.A, args.B, reply)
}

// callSum 示例：同步调用 Sum 方法
func callSum(client *rpc.Client) {
    args := &server_example.Args{A: 10, B: 20}
    var reply int
    log.Printf("Client calling Arith.Sum with A=%d, B=%d", args.A, args.B)
    err := client.Call("Arith.Sum", args, &reply)
    if err != nil {
        log.Fatalf("Error calling Arith.Sum: %v", err)
    }
    fmt.Printf("Arith: %d + %d = %d\n", args.A, args.B, reply)
}

// callAsyncMultiply 示例：异步调用 Multiply 方法
func callAsyncMultiply(client *rpc.Client) {
    args := &server_example.Args{A: 12, B: 3}
    var reply int
    log.Printf("Client initiating asynchronous call to Arith.Multiply with A=%d, B=%d", args.A, args.B)

    // client.Go() 返回一个 *rpc.Call 结构体，其中包含一个 Done 字段，是一个 channel
    call := client.Go("Arith.Multiply", args, &reply, nil) // 最后一个参数是 channel，nil表示使用默认channel

    // 可以在这里执行其他操作，不阻塞等待 RPC 结果
    fmt.Println("Client continues to do other work while RPC is in progress...")
    time.Sleep(50 * time.Millisecond) // 模拟其他工作

    // 等待 RPC 调用完成
    <-call.Done
    if call.Error != nil {
        log.Fatalf("Error during asynchronous Arith.Multiply call: %v", call.Error)
    }
    fmt.Printf("Arith (Async): %d * %d = %d\n", args.A, args.B, reply)
}

// sendMessageToMultipleHosts 示例：向多个主机发送消息
func sendMessageToMultipleHosts(hostAddresses []string) {
    fmt.Println("\n--- Sending messages to multiple hosts ---")
    var wg sync.WaitGroup

    for i, addr := range hostAddresses {
        wg.Add(1)
        go func(hostAddr string, index int) {
            defer wg.Done()
            log.Printf("Attempting to connect to host: %s", hostAddr)
            client, err := rpc.DialHTTP("tcp", hostAddr)
            if err != nil {
                log.Printf("Could not connect to host %s: %v", hostAddr, err)
                return
            }
            defer client.Close()

            args := &server_example.Args{A: index + 1, B: 10}
            var reply int
            log.Printf("Client sending message to %s: Arith.Multiply with A=%d, B=%d", hostAddr, args.A, args.B)
            err = client.Call("Arith.Multiply", args, &reply)
            if err != nil {
                log.Printf("Error calling Arith.Multiply on %s: %v", hostAddr, err)
                return
            }
            fmt.Printf("Received acknowledgement from %s: %d * %d = %d\n", hostAddr, args.A, args.B, reply)
        }(addr, i)
    }
    wg.Wait()
    fmt.Println("--- All messages sent to multiple hosts (or attempted) ---")
}