Golang高效网络通信支持双向调用通知广播插件库arpc的使用

ARPC是一个高效的Golang网络通信库，支持双向调用和通知广播功能。下面我将详细介绍其使用方法和示例代码。

特性

• 双向调用（客户端到服务端，服务端到客户端）
• 双向通知
• 同步和异步调用
• 同步和异步响应
• 批量写入（Writev/net.Buffers）
• 广播功能
• 中间件支持
• 发布/订阅模式
• OpenTracing支持

安装

1. 获取并安装arpc：

$ go get -u github.com/lesismal/arpc

1. 在代码中导入：

import "github.com/lesismal/arpc"

快速开始

服务端示例

package main

import "github.com/lesismal/arpc"

func main() {
    server := arpc.NewServer()

    // 注册路由
    server.Handler.Handle("/echo", func(ctx *arpc.Context) {
        str := ""
        if err := ctx.Bind(&str); err == nil {
            ctx.Write(str)
        }
    })

    server.Run("localhost:8888")
}

客户端示例

package main

import (
    "log"
    "net"
    "time"
    "github.com/lesismal/arpc"
)

func main() {
    client, err := arpc.NewClient(func() (net.Conn, error) {
        return net.DialTimeout("tcp", "localhost:8888", time.Second*3)
    })
    if err != nil {
        panic(err)
    }
    defer client.Stop()

    req := "hello"
    rsp := ""
    err = client.Call("/echo", &req, &rsp, time.Second*5)
    if err != nil {
        log.Fatalf("Call failed: %v", err)
    } else {
        log.Printf("Call Response: \"%v\"", rsp)
    }
}

API示例

注册路由

var handler arpc.Handler

// 包级别
handler = arpc.DefaultHandler
// 服务端
handler = server.Handler
// 客户端
handler = client.Handler

// 消息默认在同一个连接读取goroutine中顺序处理
handler.Handle("/route", func(ctx *arpc.Context) { ... })
handler.Handle("/route2", func(ctx *arpc.Context) { ... })

// 使消息由新的goroutine处理
async := true
handler.Handle("/asyncResponse", func(ctx *arpc.Context) { ... }, async)

路由中间件

import "github.com/lesismal/arpc/extension/middleware/router"

var handler arpc.Handler

handler.Use(router.Recover())
handler.Use(router.Logger())
handler.Use(func(ctx *arpc.Context) { ... })
handler.Handle("/echo", func(ctx *arpc.Context) { ... })
handler.Use(func(ctx *arpc.Context) { ... })

客户端调用方式

1. 同步调用（阻塞，带超时/上下文）

request := &Echo{...}
response := &Echo{}
timeout := time.Second*5
err := client.Call("/call/echo", request, response, timeout)
// 或使用上下文
// ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// defer cancel()
// err := client.CallWith(ctx, "/call/echo", request, response)

1. 异步调用（非阻塞，带回调和超时/上下文）

request := &Echo{...}

timeout := time.Second*5
err := client.CallAsync("/call/echo", request, func(ctx *arpc.Context) {
    response := &Echo{}
    ctx.Bind(response)
    ...    
}, timeout)

1. 通知（与CallAsync类似，但没有回调）

data := &Notify{...}
client.Notify("/notify", data, time.Second)
// 或使用上下文
// ctx, cancel := context.WithTimeout(context.Background(), time.Second)
// defer cancel()
// client.NotifyWith(ctx, "/notify", data)

广播通知

var mux = sync.RWMutex{}
var clientMap = make(map[*arpc.Client]struct{})

func broadcast() {
    var svr *arpc.Server = ... 
    msg := svr.NewMessage(arpc.CmdNotify, "/broadcast", fmt.Sprintf("broadcast msg %d", i))
    mux.RLock()
    for client := range clientMap {
        client.PushMsg(msg, arpc.TimeZero)
    }
    mux.RUnlock()
}

发布/订阅示例

服务端

import "github.com/lesismal/arpc/extension/pubsub"

var (
    address = "localhost:8888"
    password = "123qwe"
    topicName = "Broadcast"
)

func main() {
    s := pubsub.NewServer()
    s.Password = password

    // 服务端发布消息给所有客户端
    go func() {
        for i := 0; true; i++ {
            time.Sleep(time.Second)
            s.Publish(topicName, fmt.Sprintf("message from server %v", i))
        }
    }()

    s.Run(address)
}

订阅客户端

import "github.com/lesismal/arpc/log"
import "github.com/lesismal/arpc/extension/pubsub"

func onTopic(topic *pubsub.Topic) {
    log.Info("[OnTopic] [%v] \"%v\", [%v]",
        topic.Name,
        string(topic.Data),
        time.Unix(topic.Timestamp/1000000000, topic.Timestamp%1000000000).Format("2006-01-02 15:04:05.000"))
}

func main() {
    client, err := pubsub.NewClient(func() (net.Conn, error) {
        return net.DialTimeout("tcp", address, time.Second*3)
    })
    if err != nil {
        panic(err)
    }
    client.Password = password

    // 认证
    err = client.Authenticate()
    if err != nil {
        panic(err)
    }

    // 订阅主题
    if err := client.Subscribe(topicName, onTopic, time.Second); err != nil {
        panic(err)
    }

    <-make(chan int)
}

发布客户端

import "github.com/lesismal/arpc/extension/pubsub"

func main() {
    client, err := pubsub.NewClient(func() (net.Conn, error) {
        return net.DialTimeout("tcp", address, time.Second*3)
    })
    if err != nil {
        panic(err)
    }
    client.Password = password

    // 认证
    err = client.Authenticate()
    if err != nil {
        panic(err)
    }

    for i := 0; true; i++ {
        if i%5 == 0 {
            // 发布消息给所有客户端
            client.Publish(topicName, fmt.Sprintf("message from client %d", i), time.Second)
        } else {
            // 发布消息给单个客户端
            client.PublishToOne(topicName, fmt.Sprintf("message from client %d", i), time.Second)
        }
        time.Sleep(time.Second)
    }
}

更多示例

更多使用示例可以参考官方示例代码库。

ARPC提供了丰富的功能和灵活的API，可以满足各种网络通信需求，特别是需要双向通信和广播通知的场景。