从围绕API到围绕数据-使用流式编程构建更简洁的架构

// SendStream ... 
func (d *Svc) SendStream(stream MyApi_data.ProxyDialOut_SendStreamServer) error { 
 for { 
 ... 
        data, err := stream.Recv() 
 if err != nil { 
            logrus.Errorf("recv error:%v", err) 
 return err 
 } 
 ... 
 // 对data做相关的操作 
 } 
} 

// SendStream ... 
func (d *MyApiSvc) SendStream(stream MyApi_data.ProxyDialOut_SendStreamServer) error { 
 for { 
 ... 
        data, err := stream.Recv() 
 if err != nil { 
            logrus.Errorf("recv error:%v", err) 
 return err 
 } 
 ... 
 // 对data做相关的操作 
 ... 
 // 添加一个共享的监控统计器，调用上报业务，每个api都需要改动 
        counter.Add("MyApi", 1) 
 } 
} 

import "github.com/reugn/go-streams/extension" 
type Source interface{ 
 GetSource() *extension.ChanSource 
} 

type MyApiSvc struct { 
    name     string 
    ctx      context.Context 
    ch       chan any // 就是它 
    protocol string 
} 
// GetSource 实现Source接口 
func (t *MyApiSvc) GetSource() *extension.ChanSource { 
 return extension.NewChanSource(t.ch) 
} 
type DataItem struct { 
    data    any 
    session map[string]any 
} 
// SendStream ... 
func (d *MyApiSvc) SendStream(stream MyApi_data.ProxyDialOut_SendStreamServer) error { 
 for { 
 ... 
        data, err := stream.Recv() 
 if err != nil { 
            logrus.Errorf("recv error:%v", err) 
 return err 
 } 
 ... 
 // 这里不对数据做任何处理，封装之后，直接丢到chan里 
        td := new(DataItem) 
        td.session = make(map[string]any) 
        td.session["ip"] = ip 
        td.session["trace_id"] = grand.S(8) 
        td.data = data 
        d.ch <- td 
 } 
} 

    source := getDataSource(ctx, cfg.Name) // cfg.Name == "MyApi"，通过工厂函数载入配置，获得interface `Source` 
 // 调用接口 
    source.GetSource().Via(flow.NewMap(func(i interface{}) interface{} { // 步骤1，创建日志 
 // 从用户发来的每条消息都被打散成为了数据源的一条数据 
        msg := i.(model.*DataItem) 
        traceID := msg.GetSession()["trace_id"].(string) 
 // 从数据的session中获取数据的附加信息 
        tags := map[string]interface{}{ 
 "trace_id": traceID, 
 "ip":       msg.GetSession()["ip"], 
 "name":     c.Name, 
 } 
        log := logrus.WithFields(tags) 
 // 这个步骤只是为了添加一个日志对象 
 return []any{msg, log} 
 // 使用8个协程来执行这个步骤 
 }, 8)).Via(flow.NewMap(func(i interface{}) interface{} { // 步骤2，解析数据 
        arr := i.([]any) // 这里的i是上一步骤return的数据 
        msg := arr[0].(*DataItem) 
        log := arr[1].(*logrus.Entry) 
        parser := getParser(cfg.Name) // 这个工厂函数是每种数据源的个性化处理。根据配置获取一个解析器 
 // 解析数据 
        data, err := parser(ctx, msg, c.Name, msg.GetSession()["ip"]) 
 if err != nil { 
            log.Error(err) 
 return err 
 } 
 return []any{data, log} 
 }, 8)).Via(flow.NewMap(func(i interface{}) interface{} { // 步骤3，发送数据到下个服务 
        arr,ok := i.([]any) // 这里的i，就是上一步骤return的数据 
 if !ok{ 
 return i // 如果上一步骤return的是error，则直接跳过不再解析 
 } 
        data := arr[0].(*MyApiData) // 这里的data，已经是上一步骤解析出来的数据 
        log := arr[1].(*logrus.Entry) 
 // 发数数据 
        sender := getSender(cfg.Name) // 这个工厂函数为不同的数据源分配一个发送器 
        sender.Send(qdata) 
 return i 
 }, 8)).Via(flow.NewMap(func(i interface{}) interface{} { // 步骤4，统计发送成功的数据量 
        arr, ok := i.([]any) // 这里的i，就是上一步骤return的数据 
 if ok{ 
            msg := arr[0].(*DataItem) 
            log := arr[1].(*logrus.Entry) 
 // 内部统计 
            log.Info("send success") 
            controller.TraceAfter(msg.GetSession()["ip"]) 
 } 
 return i 
 }, 8)).To(extension.NewIgnoreSink())