Go语言异步服务器框架原理和实现

Go语言类库中,有两个官方的服务器框架,一个HTTP,一个是RPC。使用这个两个框架,已经能解决大部分的问题,但是,也有一些需求,这些框架是不够的,这篇文章,我们先分析一下HTTP 和 RPC服务器的特点, 然后结合这两个服务器的特点,我实现了一个新的服务器,这个服务器非常适合客户端和服务器端有大量交互的情况。

HTTP服务器的特点:

HTTP的请求 和 响应的周期如下:

对于一个HTTP 长连接,一个请求必须等到一个响应完成后,才能进行下一个请求。这就是http协议最本质的特点,是串行化的。而这个特点保证了http协议的简洁性,一个请求中间不会插入其他的请求干扰,这样不需要去对应请求和响应。但是,同时也有个弱点,那就是不适合做大量的请求。举个实际中我们遇到的例子,我们要把大量的中国客户的订单送入英国的交易所,交易所的接口是http协议的,从中国到英国,一次http的请求到响应至少需要 300ms左右,这样一秒一个连只能发送3个,就算是开十个线程发送(接口对线程总数是有限制的),1s 也只能是30个。而最高峰的时候,我们可能1s 要发送1万个订单,那采用http协议就不能满足我们的要求了(这个可以通过fix协议解决)。

当然,http可以解决批量提交的需求,只要增加一个批量提交的接口就可以了。但是,这样的实现方式不够自然,而且增加了额外的接口。

RPC服务的特点:

PRC服务器克服了http服务器串流模型,可以并发的提交请求。请求响应的周期图如下:

RPC服务,已经可以客服http服务器的串流的劣势,可以批量提交大量的数据。在局域网的中测试,1s钟可以实现3万次左右的请求。而相同的条件下,http在局域网中,只能实现1500次左右的请求,真实环境下面,延时严重,http性能会急剧下降。在两个不同的机房中,有百兆带宽相连,实际测试rpc请求是两万次左右,http是 500次左右,而且http占用很多头部的带宽。

RPC的一个核心特点是类似一次函数调用。这样一个请求 只能 对应于 一个响应。在某些情下,这似乎是不够的。举个实际的例子,我要获取一个报价的行情数据,这个时候,类似一个MessageQueue,服务器会不断的push数据给客户端。也就是一次请求,会有多次返回,持续不断的返回。

当然,RPC的一个非常重要的优势是,你不需要知道怎么去解析数据,你可以当做网络是空气,完全像写本地调用函数一样去调用rpc的函数。

异步服务器:

因为暂时我没有很好的名字来命名这个服务器,所以暂时就叫做异步服务器吧,这个服务器的特点类似一个界面程序的消息体系。我们不断的吧鼠标键盘等各种事件提交给界面程序,界面程序根据消息的类型,参数做出相应的处理。所以,我们就叫做异步服务器吧。经典的金融服务器都是异步服务器,处理机制都类似界面的消息循环机制,比如国内期货最常用的ctp交易系统,还有就是银行间,交易所和银行之间,经常用的一个协议叫做 fix,也是这样的架构。请求是一种消息,响应也是一种消息。请求响应的时序图如下:

msg1 请求之后,有两个响应,Resp1 , resp2,

msg2 有一个响应 resp3.

借鉴了rpc的特点,请求和响应都自动编码,写服务器不再为编码而烦恼,同时也不需要为是否要压缩而头痛。现在提供三种方式,gob , json, protocolbuffer. 并且可以 设置是否启用压缩的,以及压缩的格式。我

们把客户端和服务器的交互抽象为一个消息系统,先来看看客户端客户端调用

   1: client, err := NewClient("http://localhost:8080", jar, "gob", "gzip")
   2: if err != nil {
   3:     log.Println(err)
   4:     return
   5: }
   6: defer client.Close()
   7: req := NewRequest("hello", "jack", func(call *Call, status int) {
   8:     log.Println(call, call.Resp, status)
   9: })
  10: client.Go(req)
  11: req2 := NewRequest("hello", "fuck", func(call *Call, status int) {
  12:     log.Println(call, call.Resp, status)
  13: })
  14: client.Go(req2)
  15: //wait for all req is done
  16: client.Wait()

1-6行,我们建立了一个到服务器的连接,注意,我们这个服务器底层是用http包实现的。jar 是用来管理session的,这里暂时忽略,gob是编码,gzip是压缩格式。可以动态设置各种编码和压缩格式。

7-13行,NewRequest 的第一个参数是消息的类型(我建议再后面的版本中,改成NewMessage, Client.GO 改成 client.Send),叫做hello, 详细类型为了方便查看也打印,我采用字符串的格式。后面是消息的参数,可以是任何的go的结构,变量。每个请求对应一个回调函数,处理响应的消息,响应的消息保存在 call.Resp 里面,如果status == StatusDone , 表示请求结束了,服务器不会响应任何消息了,status == StatusUpdate ,说明,还会有下一个消息过来。

16行 Wait函数,其实就是一个消息循环函数,不断的从服务器端读取消息,对应到某个请求的回调函数里面。类似event loop

我们在Client里面加入心跳函数,保证能检查到链接损坏的情况,如果连接损坏,会自动结束消息循环,错误处理是一个服务器非常重要的一环。

然后我们再来看看服务器端的实现:

   1: func helloWorld(w *ResponseWriter, r *Request) {
   2:     resp := w.Resp
   3:     resp.MsgType = MsgTString
   4:     //表示我已经没有其他数据包了,这个请求已经结束了
   5:     resp.Done = true
   6:     //向客户端发送请求
   7:     w.WriteResponse(resp, "hello: " + r.GetBody().(string))
   8: }

第7行中,r.GetBody() 获取的到是上面NewRequest 中的第二个参数。

这样就是一个最简单的hello world 程序。要实现一个实战有用的服务器,的细节当然还有很多,主要的是流量控制。比如,一个用户写错程序了,错误的发起了10万个请求,服务器端不能开个10万个go进行处理,这样的话,会直接拖垮服务器,我们给每个用户设置了一个并发处理数目,最多这个用户可以并发处理多少个请求。还有一个比较重要的,对服务器来说,就是服务器服务的量的限制。我们会实时监控 cpu 内存,io的使用情况,当发现使用到某个限额的时候,服务会拒绝接受连接(事先要对性能进行测试)这些都是为了防止服务器过载 ,而实际中的服务器,这个问题其实是很常见的。

实例:可靠消息通知系统。

可靠消息通知系统实际上是一个非常常见的系统。最常用的一个例子就是数据库的master slave 模式。master里面的事件要非常可靠的通知到slave,中间不能有任何的丢失。还有一种比如交易系统中,我们会调用银行或者交易所的接口,银行在交易成功后会给我们一个通知,这个通知的消息必须可靠的被通知到目标,不能有任何的丢失。在我们的系统中,行情数据的复制也是不能有任何数据丢失的情景,为了保证A 服务器 和 B服务器有相同的行情,在从A服务器的消息要被B服务器准确的接收。当然,你也可以做一个聊天系统,这个聊天系统不会丢失任何消息。

那么如何实现这个系统呢,首先,为了保证不在内存中丢失消息,那么消息必须写盘,并且为了检测消息是否丢失,必须给消息编号。消息写盘也可以用我们开发的事务日志系统,如果消息非常的大量,那么还需要批量提交模式(Group Commit)。大部分情况下,消息丢失不是因为服务器崩溃,而且网络意外中断,这些中断往往时间很短,在1分钟以内,所以,有必要在内存中缓存部分的消息,如果网络中断,客户端再次请求时,发送当时的消息序号,这样就可以补全网络中断丢失的数据。如果时间太长了,内存中的数据不够补了,那么首先要从消息源数据库中下载历史消息,然后再接受实时的消息。整体的思路就是这样的,在这里,我们就看看我们的消息通知系统的实时广播部分的设计。

1. 消息广播基本流程: 订阅 –> 广播:

首先客户端向服务器说明,我要订阅哪些消息,比如,master slave 中,我只要写消息就好了,读消息就不需要了。然后,再向服务器请求数据,服务器广播数据给我们。注意,我们这里把订阅 和 广播分成两个部分,两个请求,那么怎么知道这两个请求是同一个人发出的呢?或者,怎么关联起来呢?这里,我用了一个session的概念,订阅的时候,把订阅的消息类型保存到session,广播的时候,从session中读取消息类型,然后发送对应的数据。

这部分的代码如下:

   1: var bmu sync.Mutex
   2: var defaultBroadcast = make(map[int64]*Broadcast)
   3: var ErrNotRingItemer = errors.New("ErrNotRingItemer")
   4: //基本上可以保证有1个小时的数据
   5: const btickSize = 3600 * 4
   6: //可以传递任意的数据
   7:  
   8: func GetBroadcast(name int64, n int) (*Broadcast, error) {
   9:     bmu.Lock()
  10:     defer bmu.Unlock()
  11:     b, ok := defaultBroadcast[name]
  12:     if ok {
  13:         return b, nil
  14:     }
  15:     b , err := NewBroadcast(name, n)
  16:     if err != nil {
  17:         return nil, err
  18:     }
  19:     defaultBroadcast[name] = b
  20:     return b, nil
  21: }
  22:  
  23: type Broadcast struct {
  24:     mu sync.RWMutex
  25:     targets map[int64]*Subscribe
  26:     ringbuffer *algo.RingBuffer
  27:     name int64
  28: }
  29:  
  30: func NewBroadcast(name int64, n int) (*Broadcast, error) {
  31:     b := &Broadcast{}
  32:     b.targets = make(map[int64]*Subscribe)
  33:     b.ringbuffer = algo.NewRingBuffer(n, nil)
  34:     b.name = name
  35:     return b, nil
  36: }
  37:  
  38: func (b *Broadcast) GetName() int64 {
  39:     return b.name
  40: }
  41:  
  42: func (b *Broadcast) Sub(id int64, req *Subscribe) {
  43:     b.mu.Lock()
  44:     defer b.mu.Unlock()
  45:     b.targets[id] = req
  46: }
  47:  
  48: func (b *Broadcast) Unsub(id int64) {
  49:     b.mu.Lock()
  50:     defer b.mu.Unlock()
  51:     delete(b.targets, id)
  52: }
  53:  
  54: //是否在buffer内部
  55: func (b *Broadcast) InBuffer(start int64, end int64) (bool, error) {
  56:     return b.ringbuffer.InBuffer(start, end)
  57: }
  58:  
  59: func (b *Broadcast) Query(start int64, end int64, ty int64) (algo.Iterator, error) {
  60:     find := &algo.RingFind{start, end, ty}
  61:     return b.ringbuffer.Find(find, true) //模糊查找,不是精确匹配
  62: }
  63:  
  64: //如果要提供查询功能,那么就要缓存数据,一般采用ringbuffer
  65: //data要满足下面的条件:
  66: //1. 存在一个递增着的ID
  67: //2. 实现BufferItemer接口
  68: func (b *Broadcast) Push(item algo.RingItemer) error {
  69:     b.mu.RLock()
  70:     defer b.mu.RUnlock()
  71:     item2, err := b.ringbuffer.Push(item)
  72:     if err != nil {
  73:         return err
  74:     }
  75:     for _, v := range b.targets {
  76:         //过滤不想发送的
  77:         if (v.Check(b.name, item2.Type)) {
  78:             v.Send(item)
  79:         }
  80:     }
  81:     return nil
  82: }
  83:  
  84: func (b *Broadcast) Find(find *algo.RingFind) (algo.Iterator, error) {
  85:     return b.ringbuffer.Find(find, true)
  86: }
  87:  
  88: type Subscribe struct {
  89:     mu sync.Mutex
  90:     ch chan interface{}
  91:     tys map[int64]int64
  92: }
  93:  
  94: func NewSubscribe(n int) (*Subscribe) {
  95:     s := &Subscribe{}
  96:     s.ch = make(chan interface{}, n)
  97:     s.tys = make(map[int64]int64)
  98:     return s
  99: }
 100:  
 101: func (s *Subscribe) Add(bname int64, ty int64) {
 102:     s.mu.Lock()
 103:     defer s.mu.Unlock()
 104:     s.tys[bname] = ty
 105: }
 106:  
 107: func (s *Subscribe) Check(bname int64, dataty int64) bool {
 108:     s.mu.Lock()
 109:     defer s.mu.Unlock()
 110:     ty, ok := s.tys[bname]
 111:     if !ok { //没有订阅
 112:         return false
 113:     }
 114:     if ty == algo.AnyType || dataty == ty {
 115:         return true
 116:     }
 117:     return false
 118: }
 119:  
 120: func (s *Subscribe) Read(buf []interface{}) (int) {
 121:     var i = 1
 122:     buf[0] = <-s.ch
 123:     for {
 124:         if i == len(buf) {
 125:             return i
 126:         }
 127:         select {
 128:         case data := <-s.ch:
 129:             buf[i] = data
 130:             i++
 131:         default:
 132:             return i
 133:         }
 134:     }
 135:     panic("nerver reach")
 136: }
 137:  
 138: func (s *Subscribe) Send(data interface{}) {
 139:      select {
 140:      case s.ch <- data :
 141:      default:
 142:          //清除旧的数据
 143:          s.Clear()
 144:          //发送结束标志位
 145:          s.ch <- nil
 146:      }
 147: }
 148:  
 149: func (s *Subscribe) Clear() {
 150:     for {
 151:         select {
 152:         case <-s.ch:
 153:         default:
 154:             return
 155:         }
 156:     }
 157: }
 158:  

这里,有个数据结构叫做RingBuffer, 是一个环状的buffer,非常适合做缓存固定数目的数据,用于广播。广播是用管道来传输数据的,管道的性能实际上已经非常的高,不需要什么无锁队列之类的。在这里也给管道加上buffer使得,消息意外的扰动,不会使得带宽不够用而立马堵塞。

2. 接受消息:

在用户登录后,如果有权限,那么就可以作为消息源客户端,消息源的代码如下:

   1: func pushTick(w *asyn.ResponseWriter, r *asyn.Request) {
   2:     event := r.GetBody().(*response.OrderBookEvent)
   3:     b, _ := GetBroadcast(event.InstrumentId, btickSize)
   4:     b.Push(event)
   5:     asyn.Log().Println(event)
   6:     asyn.OKHandle(w, r)
   7: }

第2行: 从请求中获取 消息事件。

第3行: event.InstrumentId 是消息的类型,btickSzie 是缓存的数据数目。

第6行: 向客户端发送OK,确认消息发送成功。

每个消息是否发送成功,都有确认。这样,客户端就知道上次消息发送到哪里了。

3. 订阅:

   1: func subscribe(w *asyn.ResponseWriter, r *asyn.Request) {
   2:     instId := r.GetBody().(int64)
   3:     log.Println("sub", instId)
   4:     b, err := GetBroadcast(instId, btickSize)
   5:     if err != nil {
   6:         r.SetErr(err)
   7:         asyn.ErrorHandle(w, r)
   8:         return
   9:     }
  10:     //订阅的size
  11:     //get and set 要成为一个原子操作
  12:     session := r.GetSession()
  13:     session.Get3("subscribe", func (data interface{}) interface{} {
  14:         if data == nil {
  15:             data = NewSubscribe(4096)
  16:         }
  17:         sub := data.(*Subscribe)
  18:         //广播, 类型
  19:         id := int64(uintptr(unsafe.Pointer(session)))
  20:         sub.Add(instId, algo.AnyType)
  21:         b.Sub(id, sub)
  22:         session.OnDelete(func () {
  23:             b.Unsub(id)
  24:         })
  25:         return sub
  26:     })
  27:     asyn.OKHandle(w, r)
  28: }

第2行:获取消息的类型,通过这个类型,可以找到对应的广播对象。

第12-30行:这是一个线程安全的session操作,具体看一下session.Get3 的实现就知道了:

   1: func (s *Session) Get3(name string, callback func (interface{}) interface{}) interface{} {
   2:     s.mu.Lock()
   3:     defer s.mu.Unlock()
   4:     data, err := s.get(name)
   5:     if err != nil {
   6:         data = nil
   7:     }
   8:     data = callback(data)
   9:     s.set(name, data)
  10:     return data
  11: }

s.get 获取session的数据,如果没有session数据,那么为nil。简单的说,这里的意思是:如果session “subscribe” 如果还没有设置,那么就新建一个对象,如果已经设置了,那么读取这个对象,并且,这个操作是线程安全的。

这里还添加了一个session撤销时候的操作。

4. 广播:

   1: //读取广播数据
   2: func read(w *asyn.ResponseWriter, r *asyn.Request) {
   3:     session := r.GetSession()
   4:     //从session 中获取subscribe 对象
   5:     sub := session.Get3("subscribe", func (data interface{}) interface{} {
   6:         if data == nil {
   7:             data = NewSubscribe(4096)
   8:         }
   9:         return data
  10:     }).(*Subscribe)
  11:     depth := r.GetBody().(int)
  12:     log.Println("get subscribe")
  13:     resp := w.Resp
  14:     if depth == 0 {
  15:         resp.MsgType = "ticks"
  16:     } else {
  17:         resp.MsgType = "ticks1"
  18:     }
  19:     buf := make([]interface{}, 1024)
  20:     dg := make([]*response.OrderBookEvent, 1024)
  21:     tick1 := make([]*base.TickGo, 1024)
  22:     for {
  23:         n := sub.Read(buf)
  24:         for i := 0; i < n; i++ {
  25:             if buf[i] == nil {
  26:                 //close by broadcast
  27:                 r.SetErr(errors.New("501"))
  28:                 asyn.ErrorHandle(w, r)
  29:                 return
  30:             }
  31:             if depth == 0 {
  32:                 dg[i] = buf[i].(*response.OrderBookEvent)
  33:             } else {
  34:                 tick1[i] = buf[i].(*response.OrderBookEvent).ToTickGo()
  35:             }
  36:         }
  37:         var err error
  38:         if depth == 0 {
  39:             err = w.WriteResponse(resp, dg[:n])
  40:         } else {
  41:             err = w.WriteResponse(resp, tick1[:n])
  42:         }
  43:         if err != nil {
  44:             r.SetErr(err)
  45:             asyn.ErrorHandle(w, r)
  46:             return
  47:         }
  48:     }
  49: }

read 有个depth参数,这是行情的深度。股票期货里面都有后这个概念。传说中的几档行情。

第26行:这里有个close。一般来说,是因为网络拥堵 或者 异常,无法发送数据了。

还有一点要注意,这里的行情是批量发送的。sub.Read 尽可能多的读取数据,减少网络io的次数。

当然,服务器框架本身提供了心跳机制,对消息广播系统,实时性是非常重要的,即时的检查出网络异常,才能保证实时性。

以上是对我们的异步消息服务器框架的一个简单的介绍。设计这框架,非常重要的两个理念:

1. 模块化的设计,一个功能,就对应一个函数。

2. 模块之间的通讯采用session,而对于比较复杂的通讯,可以自己建立一个线程安全的数据结构,比如这里的Broadcast 和 Subscribe

原文发布于微信公众号 - Golang语言社区(Golangweb)

原文发表时间:2016-07-28

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

发表于

我来说两句

0 条评论
登录 后参与评论

相关文章

来自专栏深度学习计算机视觉

操作系统知识梳理共9次缺页

第一章:概述 什么是操作系统? 是一段一直运行在计算机上的程序 是资源的分配者 向上管理软件向下管理硬件 为用户提供良好接口 中断的概念? 中断指当出现需要时,...

28550
来自专栏嵌入式程序猿

【速来围观】IAR环境下代码覆盖率功能的使用

在设计测试程序,验证是否所有的代码都被执行到时,就要考虑到代码覆盖率,IAR环境下的代码覆盖率是一个在这方面很有用的功能,且使用方便,今天我们就来讲讲这一功能如...

44460
来自专栏Hadoop和Spark

用AWS部署一个无服务架构的个人网站

在这篇文章里我想介绍下怎样利用AWS(hjlouyoujuqi360com)部署一个无服务架构的个人网站。这个个人网站将具备以下特点:

47340
来自专栏FreeBuf

Grub2被曝登陆验证绕过0Day,影响众多Linux版本(CVE-2015-8370)

描述 近日,研究人员发现了一个Grub2的漏洞,版本1.98(2009年发布)到2.02(2015年发布)均受影响。本地用户能够通过这个漏洞绕过任何形式的认证(...

28060
来自专栏田京昆的专栏

基于hashicorp/raft的分布式一致性实战教学

hashicorp/raft是raft算法的一种比较流行的golang实现,基于它能够比较方便的构建具有强一致性的分布式系统。本文通过实现一个简单的分布式缓存系...

2.3K150
来自专栏HaHack

化繁为简的企业级 Git 管理实战(五):二进制大文件的版本控制

19170
来自专栏逆向技术

Win32之内存管理之虚拟内存跟物理内存

  我们知道每个应用程序都有自己独立的4GB空间.  假设A进程的 地址123 存储了10  那么B进程的123地址 存储了20

19640
来自专栏pangguoming

Git可视化教程——Git Gui的使用

在Git简介一文中已经对Git进行了简单的介绍,但是理论知识过于枯燥,加上本人专业知识不够扎实,使得初学者在Git的使用上还是会有很大的困难。虽然我更推荐使用G...

33930
来自专栏玄魂工作室

CTF实战4 HTTP协议及嗅探抓包

HTTP(Hypertext Transfer Protocol)中文<超文本传输协议>,是一种为分布式,合作式,多媒体信息系统服务,面向应用层的协议,是Int...

12820
来自专栏用户画像

3.1.4.1 基本分页存储管理方式

非连续分配允许一个程序分散地装入到不相邻的内存分区中,根据分区的大小是否固定分为分页存储管理方式和分段存储管理方式。

9010

扫码关注云+社区

领取腾讯云代金券