go网络库cellent实现socket聊天功能

OStack程序员社区-中国程序员成长平台 › 门户 › 编程› Go›Go教程

原作者: [db:作者] 来自: [db:来源] 收藏邀请

一、介绍

cellnet是一个组件化、高扩展性、高性能的开源服务器网络库

git地址：https://github.com/davyxu/cellnet

主要使用领域：

游戏服务器

方便定制私有协议，快速构建逻辑服务器、网关服务器、服务器间互联互通、对接第三方SDK、转换编码协议等
ARM设备

设备间网络通讯
证券软件

内部RPC

支持多种传输协议：

TCP

TCP连接器的重连，侦听器的优雅重启。
UDP

纯UDP裸包收发
HTTP(测试中)

侦听器的优雅重启, 支持json及form的收发及封装。编写游戏服务器时，不再需要使用第三方HTTP服务器对接SDK。

注: 如需要对接大规模网页, 请使用第三方专业网络库,如(https://github.com/gin-gonic/gin), cellnet的http支持主要目的在用统一的地址及peer管理
WebSocket

采用(github.com/gorilla/websocket)实现

支持混合消息编码：

cellnet内建支持以下数据编码:
- Google Protobuf (https://github.com/google/protobuf)
- json 适合与第三方服务器通信
- 二进制协议(https://github.com/davyxu/goobjfmt)
  
  内存流直接序列化, 适用于服务器内网传输
- ProtoPlus(https://github.com/davyxu/protoplus)
  
  增加并优化过的Protobuf的编码格式
可以通过codec包自行添加新的编码格式
支持混合编码收发

无需改动代码，只需调整消息注册方式，即可达成运行期同时收发不同编码的封包
- 与其他语言编写的服务器使用protobuf
- 与web服务器使用json通信
- 与Unity3D(C#)使用ProtoPlus(github.com/davyxu/protoplus)协议通信
优点：
- 享受不同通信协议格式的优点，避免缺点。如Protobuf适合服务器间的传输，封包小，适应面广。
- 私有的二进制协议方便加密和客户端处理，低GC和实现复杂度。

支持多线程和单线程设计架构：

单线程异步逻辑，适用于MMORPG复杂交互，免加锁处理共享数据。
多线程同步逻辑，适用于机器人逻辑，每个机器人使用独立的goroutine收发处理，机器人间互不干扰。
多线程并发逻辑，适用于网关，消息转发，HTTP服务器，每条消息处理都会在完全并发下。

支持远程过程调用（RPC）：

支持同步RPC，适用于后台服务器向其他服务器请求数据后再顺处理事务。
支持异步RPC，适用于单线程服务器逻辑。

cellnet的主要流程和架构：

socket连接管理：cellnet网络库使用连接器和接受器（connector，acceptor）管理socket连接
会话（session）：客户端和服务器连接使用会话（session）处理收发包流程，收发包的流程将事件通过事件的回调（cellnet.Eventfunc）派发
包处理（packet）：cellent中的packet包处理会话收发流程派发的事件，实现变长封包的解析，处理和收发
编码器：用户的封包使用编码器（codec）负责原始封包的字节数组和用户消息间的转换
消息队列：可以将收到的消息按顺序排队并提供给用户进行处理
消息元信息（messageMeta）：为所有的系统提供静态的消息扩展信息，如消息的ID，编码器，创建方法等

二、客户端功能

#官方demo
package main

import (
    "bufio"
    "github.com/davyxu/cellnet"
    "github.com/davyxu/cellnet/examples/chat/proto"
    "github.com/davyxu/cellnet/peer"
    "github.com/davyxu/cellnet/proc"
    "github.com/davyxu/golog"
    "os"
    "strings"

    _ "github.com/davyxu/cellnet/peer/tcp"
    _ "github.com/davyxu/cellnet/proc/tcp"
)

var log = golog.New("client")

func ReadConsole(callback func(string)) {

    for {

        // 从标准输入读取字符串，以\n为分割
        text, err := bufio.NewReader(os.Stdin).ReadString('\n')
        if err != nil {
            break
        }

        // 去掉读入内容的空白符
        text = strings.TrimSpace(text)

        callback(text)

    }

}

func main() {

    // 创建一个事件处理队列，整个客户端只有这一个队列处理事件，客户端属于单线程模型
    queue := cellnet.NewEventQueue()

    // 创建一个tcp的连接器，名称为client，连接地址为127.0.0.1:8801，将事件投递到queue队列,单线程的处理（收发封包过程是多线程）
    p := peer.NewGenericPeer("tcp.Connector", "client", "127.0.0.1:18801", queue)

    // 设定封包收发处理的模式为tcp的ltv(Length-Type-Value), Length为封包大小，Type为消息ID，Value为消息内容
    // 并使用switch处理收到的消息
    proc.BindProcessorHandler(p, "tcp.ltv", func(ev cellnet.Event) {
        switch msg := ev.Message().(type) {
        case *cellnet.SessionConnected:
            log.Debugln("client connected")
        case *cellnet.SessionClosed:
            log.Debugln("client error")
        case *proto.ChatACK:
            log.Infof("sid%d say: %s", msg.Id, msg.Content)
        }
    })

    // 开始发起到服务器的连接
    p.Start()

    // 事件队列开始循环
    queue.StartLoop()

    log.Debugln("Ready to chat!")

    // 阻塞的从命令行获取聊天输入
    ReadConsole(func(str string) {

        p.(interface {
            Session() cellnet.Session
        }).Session().Send(&proto.ChatREQ{
            Content: str,
        })

    })

}

三、服务端功能

#官方demo
package main

import (
    "github.com/davyxu/cellnet"
    "github.com/davyxu/cellnet/examples/chat/proto"
    "github.com/davyxu/cellnet/peer"
    "github.com/davyxu/cellnet/proc"
    "github.com/davyxu/golog"

    _ "github.com/davyxu/cellnet/peer/tcp"
    _ "github.com/davyxu/cellnet/proc/tcp"
)

var log = golog.New("server")

func main() {

    // 创建一个事件处理队列，整个服务器只有这一个队列处理事件，服务器属于单线程服务器
    queue := cellnet.NewEventQueue()

    // 创建一个tcp的侦听器，名称为server，连接地址为127.0.0.1:8801，所有连接将事件投递到queue队列,单线程的处理（收发封包过程是多线程）
    p := peer.NewGenericPeer("tcp.Acceptor", "server", "127.0.0.1:18801", queue)

    // 设定封包收发处理的模式为tcp的ltv(Length-Type-Value), Length为封包大小，Type为消息ID，Value为消息内容
    // 每一个连接收到的所有消息事件(cellnet.Event)都被派发到用户回调, 用户使用switch判断消息类型，并做出不同的处理
    proc.BindProcessorHandler(p, "tcp.ltv", func(ev cellnet.Event) {

        switch msg := ev.Message().(type) {
        // 有新的连接
        case *cellnet.SessionAccepted:
            log.Debugln("server accepted")
        // 有连接断开
        case *cellnet.SessionClosed:
            log.Debugln("session closed: ", ev.Session().ID())
        // 收到某个连接的ChatREQ消息
        case *proto.ChatREQ:

            // 准备回应的消息
            ack := proto.ChatACK{
                Content: msg.Content,       // 聊天内容
                Id:      ev.Session().ID(), // 使用会话ID作为发送内容的ID
            }

            // 在Peer上查询SessionAccessor接口，并遍历Peer上的所有连接，并发送回应消息（即广播消息）
            p.(cellnet.SessionAccessor).VisitSession(func(ses cellnet.Session) bool {

                ses.Send(&ack)

                return true
            })

        }

    })

    // 开始侦听
    p.Start()

    // 事件队列开始循环
    queue.StartLoop()

    // 阻塞等待事件队列结束退出( 在另外的goroutine调用queue.StopLoop() )
    queue.Wait()

}