GoFrame 連接對象-消息包處理

?gtcp?提供了許多方便的原生操作連接數據的方法，但是在絕大多數的應用場景中，開發(fā)者需要自己設計數據結構，并進行封包/解包處理，由于?TCP?消息協議是沒有消息邊界保護的，因此復雜的網絡通信環(huán)境中很容易出現粘包的情況。因此?gtcp?也提供了簡單的數據協議，方便開發(fā)者進行消息包交互，開發(fā)者不再需要擔心消息包的處理細節(jié)，包括封包/解包處理，這一切復雜的邏輯?gtcp?已經幫你處理好了。

簡單協議

?gtcp?模塊提供了簡單輕量級數據交互協議，效率非常高，協議格式如下：

數據長度(16bit)|數據字段(變長)

1. 數據長度：默認為16位(2字節(jié))，用于標識該消息體的數據長度，單位為字節(jié)，不包含自身的2字節(jié)；
2. 數據字段：變長，根據數據長度可以知道，數據最大長度不能超過?0xFFFF = 65535 bytes = 64 KB?；

簡單協議由?gtcp?封裝實現，如果開發(fā)者客戶端和服務端如果都使用?gtcp?模塊來通信則無需關心協議實現，專注數據字段封裝/解析實現即可。如果涉及和其他開發(fā)語言對接，則需要按照該協議實現對接即可，由于簡單協議非常簡單輕量級，因此對接成本很低。

數據字段也可以為空，即沒有任何長度。

操作方法

https://pkg.go.dev/github.com/gogf/gf/v2/net/gtcp

type Conn
    func (c *Conn) SendPkg(data []byte, option ...PkgOption) error
    func (c *Conn) SendPkgWithTimeout(data []byte, timeout time.Duration, option ...PkgOption) error
    func (c *Conn) SendRecvPkg(data []byte, option ...PkgOption) ([]byte, error)
    func (c *Conn) SendRecvPkgWithTimeout(data []byte, timeout time.Duration, option ...PkgOption) ([]byte, error)
    func (c *Conn) RecvPkg(option ...PkgOption) (result []byte, err error)
    func (c *Conn) RecvPkgWithTimeout(timeout time.Duration, option ...PkgOption) ([]byte, error)

可以看到，消息包方法命名是在原有的基本連接操作方法中加上了?Pkg?關鍵詞便于區(qū)分。

其中，請求參數中的?PkgOption?數據結構如下，用于定義消息包接收策略：

// 數據讀取選項
type PkgOption struct {
	HeaderSize  int   // 自定義頭大小(默認為2字節(jié)，最大不能超過4字節(jié))
	MaxDataSize int   // (byte)數據讀取的最大包大小，默認最大不能超過2字節(jié)(65535 byte)
	Retry       Retry // 失敗重試策略
}

使用示例

示例1，基本使用

package main

import (
	"fmt"
	"github.com/gogf/gf/v2/net/gtcp"
	"github.com/gogf/gf/v2/os/glog"
	"github.com/gogf/gf/v2/util/gconv"
	"time"
)

func main() {
	// Server
	go gtcp.NewServer("127.0.0.1:8999", func(conn *gtcp.Conn) {
		defer conn.Close()
		for {
			data, err := conn.RecvPkg()
			if err != nil {
				fmt.Println(err)
				break
			}
			fmt.Println("receive:", data)
		}
	}).Run()

	time.Sleep(time.Second)

	// Client
	conn, err := gtcp.NewConn("127.0.0.1:8999")
	if err != nil {
		panic(err)
	}
	defer conn.Close()
	for i := 0; i < 10000; i++ {
		if err := conn.SendPkg([]byte(gconv.String(i))); err != nil {
			glog.Error(err)
		}
		time.Sleep(1*time.Second)
	}
}

這個示例比較簡單，執(zhí)行后，輸出結果為：

receive: [48]
receive: [49]
receive: [50]
receive: [51]
...

示例2，自定義數據結構

大多數場景下，我們需要對發(fā)送的消息能自定義數據結構，開發(fā)者可以利用數據字段傳遞任意的消息內容實現。

以下是一個簡單的自定義數據結構的示例，用于客戶端上報當前主機節(jié)點的內存及CPU使用情況，示例代碼位于：https://github.com/gogf/gf/v2/tree/master/.example/net/gtcp/pkg_operations/monitor

在該示例中，數據字段使用了?JSON?數據格式進行自定義，便于數據的編碼/解碼。

實際場景中，開發(fā)者往往需要自定義包解析協議，或者采用較通用的?protobuf?二進制包封裝/解析協議。

• ?types/types.go ?

數據結構定義。

package types


import "github.com/gogf/gf/v2/frame/g"


type NodeInfo struct {
    Cpu       float32 // CPU百分比(%)
    Host      string  // 主機名稱
    Ip        g.Map   // IP地址信息(可能多個)
    MemUsed   int     // 內存使用(byte)
    MemTotal  int     // 內存總量(byte)
    Time      int     // 上報時間(時間戳)
}

• ?gtcp_monitor_server.go ?

服務端。

package main


import (
    "encoding/json"
    "github.com/gogf/gf/v2/net/gtcp"
    "github.com/gogf/gf/v2/os/glog"
    "github.com/gogf/gf/.example/net/gtcp/pkg_operations/monitor/types"
)


func main() {
    // 服務端，接收客戶端數據并格式化為指定數據結構，打印
    gtcp.NewServer("127.0.0.1:8999", func(conn *gtcp.Conn) {
        defer conn.Close()
        for {
            data, err := conn.RecvPkg()
            if err != nil {
                if err.Error() == "EOF" {
                    glog.Println("client closed")
                }
                break
            }
            info := &types.NodeInfo{}
            if err := json.Unmarshal(data, info); err != nil {
                glog.Errorf("invalid package structure: %s", err.Error())
            } else {
                glog.Println(info)
                conn.SendPkg([]byte("ok"))
            }
        }
    }).Run()
}

• ?gtcp_monitor_client.go ?

客戶端。

package main


import (
    "encoding/json"
    "github.com/gogf/gf/v2/frame/g"
    "github.com/gogf/gf/v2/net/gtcp"
    "github.com/gogf/gf/v2/os/glog"
    "github.com/gogf/gf/v2/os/gtime"
    "github.com/gogf/gf/.example/net/gtcp/pkg_operations/monitor/types"
)


func main() {
    // 數據上報客戶端
    conn, err := gtcp.NewConn("127.0.0.1:8999")
    if err != nil {
        panic(err)
    }
    defer conn.Close()
    // 使用JSON格式化數據字段
    info, err := json.Marshal(types.NodeInfo{
        Cpu       : float32(66.66),
        Host      : "localhost",
        Ip        : g.Map {
            "etho" : "192.168.1.100",
            "eth1" : "114.114.10.11",
        },
        MemUsed   : 15560320,
        MemTotal  : 16333788,
        Time      : int(gtime.Timestamp()),
    })
    if err != nil {
        panic(err)
    }
    // 使用 SendRecvPkg 發(fā)送消息包并接受返回
    if result, err := conn.SendRecvPkg(info); err != nil {
        if err.Error() == "EOF" {
            glog.Println("server closed")
        }
    } else {
        glog.Println(string(result))
    }
}

• 執(zhí)行后

客戶端輸出結果為：

  2019-05-03 13:33:25.710 ok

服務端輸出結果為：

  2019-05-03 13:33:25.710 &{66.66 localhost map[eth1:114.114.10.11 etho:192.168.1.100] 15560320 16333788 1556861605}
  2019-05-03 13:33:25.710 client closed