chenyh
/
vRp.CD2g_test
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity
vRp.CD2g_test/server/gateway/README.md

11 KiB
Raw Blame History

Gateway

Go doc

gateway 是用于处理服务器消息的网关模块，适用于对客户端消息进行处理、转发的情况。

目录导航

列出了该 package 下所有的函数及类型定义，可通过目录导航进行快捷跳转 ❤️

展开 / 折叠目录导航

包级函数定义

函数名称 描述
NewEndpoint 创建网关端点
WithEndpointStateEvaluator 设置端点健康值评估函数
WithEndpointConnectionPoolSize 设置端点连接池大小
WithEndpointReconnectInterval 设置端点重连间隔
NewGateway 基于 server.Server 创建 Gateway 网关服务器
WithEndpointSelector 设置端点选择器
MarshalGatewayOutPacket 将数据包转换为网关出网数据包
UnmarshalGatewayOutPacket 将网关出网数据包转换为数据包
MarshalGatewayInPacket 将数据包转换为网关入网数据包
UnmarshalGatewayInPacket 将网关入网数据包转换为数据包

类型定义

类型 名称 描述
STRUCT Endpoint 网关端点
STRUCT EndpointOption 网关端点选项
STRUCT ConnectionOpenedEventHandle 暂无描述...
STRUCT EndpointSelector 暂无描述...
STRUCT Gateway 基于 server.Server 实现的网关服务器
STRUCT Option 网关选项
INTERFACE Scanner 端点扫描器

详情信息

func NewEndpoint(name string, cli *client.Client, options ...EndpointOption) *Endpoint

创建网关端点

func WithEndpointStateEvaluator(evaluator func (costUnixNano float64) float64) EndpointOption

设置端点健康值评估函数

func WithEndpointConnectionPoolSize(size int) EndpointOption

设置端点连接池大小

  • 默认为 DefaultEndpointConnectionPoolSize
  • 端点连接池大小决定了网关服务器与端点服务器建立的连接数，如果 <= 0 则会使用默认值
  • 在网关服务器中，多个客户端在发送消息到端点服务器时，会共用一个连接，适当的增大连接池大小可以提高网关服务器的承载能力

func WithEndpointReconnectInterval(interval time.Duration) EndpointOption

设置端点重连间隔

  • 默认为 DefaultEndpointReconnectInterval
  • 端点在连接失败后会在该间隔后重连，如果 <= 0 则不会重连

func NewGateway(srv *server.Server, scanner Scanner, options ...Option) *Gateway

基于 server.Server 创建 Gateway 网关服务器

func WithEndpointSelector(selector EndpointSelector) Option

设置端点选择器

  • 默认情况下，网关会随机选择一个端点作为目标，如果需要自定义端点选择器，可以通过该选项设置

func MarshalGatewayOutPacket(addr string, packet []byte) ([]byte, error)

将数据包转换为网关出网数据包

  • | identifier(4) | ipv4(4) | port(2) | packet |

func UnmarshalGatewayOutPacket(data []byte) (addr string, packet []byte, err error)

将网关出网数据包转换为数据包

  • | identifier(4) | ipv4(4) | port(2) | packet |

func MarshalGatewayInPacket(addr string, currentTime int64, packet []byte) ([]byte, error)

将数据包转换为网关入网数据包

  • | ipv4(4) | port(2) | cost(4) | packet |

func UnmarshalGatewayInPacket(data []byte) (addr string, sendTime int64, packet []byte, err error)

将网关入网数据包转换为数据包

  • | ipv4(4) | port(2) | cost(4) | packet |

Endpoint STRUCT

网关端点

  • 每一个端点均表示了一个目标服务，网关会将数据包转发到该端点，由该端点负责将数据包转发到目标服务。
  • 每个端点会建立一个连接池，默认大小为 DefaultEndpointConnectionPoolSize，可通过 WithEndpointConnectionPoolSize 进行设置。
  • 网关在转发数据包时会自行根据延迟维护端点健康值，端点健康值越高，网关越倾向于将数据包转发到该端点。
  • 端点支持连接未中断前始终将数据包转发到特定端点，这样可以保证连接的状态维持。

连接池：

  • 连接池大小决定了网关服务器与端点服务器建立的连接数，例如当连接池大小为 1 时，那么所有连接到该端点的客户端都会共用一个连接。
  • 连接池的设计可以突破单机理论 65535 个 WebSocket 客户端的限制，适当的增大连接池大小可以提高网关服务器的承载能力。
type Endpoint struct {
	gateway     *Gateway
	client      []*client.Client
	name        string
	address     string
	state       atomic.Float64
	evaluator   func(costUnixNano float64) float64
	connections *haxmap.Map[string, *server.Conn]
	rci         time.Duration
	cps         int
}

func (*Endpoint) GetName() string

获取端点名称

func (*Endpoint) GetAddress() string

获取端点地址

func (*Endpoint) GetState() float64

获取端点健康值

func (*Endpoint) Forward(conn *server.Conn, packet []byte, callback ...func (err error))

转发数据包到该端点

  • 端点在处理数据包时，应区分数据包为普通直连数据包还是网关数据包。可通过 UnmarshalGatewayOutPacket 进行数据包解析，当解析失败且无其他数据包协议时，可认为该数据包为普通直连数据包。

EndpointOption STRUCT

网关端点选项

type EndpointOption func(endpoint *Endpoint)

ConnectionOpenedEventHandle STRUCT

type ConnectionOpenedEventHandle func(gateway *Gateway, conn *server.Conn)

EndpointSelector STRUCT

type EndpointSelector func(endpoints []*Endpoint) *Endpoint

Gateway STRUCT

基于 server.Server 实现的网关服务器

  • 网关服务器是一个特殊的服务器，它会通过扫描器扫描端点列表，然后连接到端点列表中的所有端点，当端点连接成功后，网关服务器会将客户端的连接数据转发到端点服务器
  • 由于该网关为多个客户端共享一个端点的连接，所以不会受限于单机 65535 个端口的限制
  • 需要注意的是，网关可以通过扫描器得到每个端点需要建立多少个连接，该连接总数将受限与 65535 个端口的限制

一些特性：

  • 支持 server.Server 所支持的所有 Socket 网络类型
  • 支持将客户端网络类型进行不同的转换，例如：客户端使用 Websocket 连接，但是网关服务器可以将其转换为 TCP 端点的连接
  • 支持客户端消息绑定，在客户端未断开连接的情况下，可以将客户端的连接绑定到某个端点，这样该客户端的所有消息都会转发到该端点
  • 根据端点延迟实时调整端点状态评分，根据评分选择最优的端点，默认评分算法为：1 / (1 + 1.5 * ${DelaySeconds})
type Gateway struct {
	*events
	srv     *server.Server
	scanner Scanner
	es      map[string]map[string]*Endpoint
	esm     sync.Mutex
	ess     EndpointSelector
	closed  bool
	running bool
	cce     map[string]*Endpoint
	cceLock sync.RWMutex
}

func (*Gateway) Run(addr string) error

运行网关

查看 / 收起单元测试 

func TestGateway_Run(t *testing.T) {
	gw := gateway.NewGateway(server.New(server.NetworkWebsocket, server.WithDeadlockDetect(time.Second*3)), new(Scanner))
	gw.RegConnectionReceivePacketEventHandle(func(gateway *gateway.Gateway, conn *server.Conn, packet []byte) {
		endpoint, err := gateway.GetConnEndpoint("test", conn)
		if err == nil {
			endpoint.Forward(conn, packet)
		}
	})
	gw.RegEndpointConnectReceivePacketEventHandle(func(gateway *gateway.Gateway, endpoint *gateway.Endpoint, conn *server.Conn, packet []byte) {
		conn.Write(packet)
	})
	if err := gw.Run(":8888"); err != nil {
		panic(err)
	}
}

func (*Gateway) Shutdown()

关闭网关

func (*Gateway) Server() *server.Server

获取网关服务器核心

func (*Gateway) GetEndpoint(name string) ( *Endpoint, error)

获取一个可用的端点

  • name: 端点名称

func (*Gateway) GetConnEndpoint(name string, conn *server.Conn) ( *Endpoint, error)

获取一个可用的端点，如果客户端已经连接到了某个端点，将优先返回该端点

  • 当连接到的端点不可用或没有连接记录时，效果同 GetEndpoint 相同
  • 当连接行为为有状态时，推荐使用该方法

func (*Gateway) SwitchEndpoint(source *Endpoint, dest *Endpoint)

将端点端点的所有连接切换到另一个端点

Option STRUCT

网关选项

type Option func(gateway *Gateway)

Scanner INTERFACE

端点扫描器

type Scanner interface {
	GetEndpoints() ([]*Endpoint, error)
	GetInterval() time.Duration
}

func (*Scanner) GetEndpoints() ( []*gateway.Endpoint, error)

func (*Scanner) GetInterval() time.Duration