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

   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: 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: }

   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:  

   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: }

   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: }

   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: }

   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: }