使用WaitGroup, Channel和Context打造一个并发用户标签查询器

[kevinyan815]

这个例子是使用WaitGroup编排多个goroutine并发完成用户多个标签数据的查询工作。至于每个标签应该怎么查询则由具体的标签查询实现类型实现Picker接口的PickTagValueForUser方法，在方法里实现自己标签的查询逻辑。

下面的Picker接口约定了每个标签查询器必须要实现的两个方法

type Picker interface {
    // 用于查询用户的标签值
    PickTagValueForUser (userId int64, args ... interface{})
    // 通知查询到的标签值
    Notify () <-chan interface{}
}

resolveTagPicker函数用于通过标签名解析出每个对应标签查询器对象。

// 根据标签名解析出对应的TagPicker
func resolveTagPicker (tagName string) Picker {
    switch tagName {
    case TAG_ORDER_NUM: // 这是个常量 值是字符串order_num 
        return &OrderNumTagPicker{
            TagName: tagName,
            ValueCh: make(chan interface{}),
        }
    default:
        return nil
    }
}

下面是对外提供的对多个用户标签进行并发查询的查询方法，BulkQueryUserTagValue会根据要查询的多个用户标签每个都开启一个goroutine 执行queryTagValue方法。 queryTagValue方法就是每个标签查询器执行的goroutine，它会根据标签标识实例化出相应的标签查询器，然后再开启一个goroutine执行标签查询器实现的PickTagValueForUser方法，查询到后会通过标签查询器的Channel获得标签值。这里再开一个goroutine去执行PickTagValueForUser方法的原因是要做好查询器的超时处理。 queryTagValue会同时接收标签查询器查询到的结果值（通过Notify方法返回的Channel）和ctx.Done() 这个Channel。 如果在ctx.Done()通道接收到值时还没有从查询器的Channel接收到标签值，则视为超时。

// 对外提供的批量查询用户标签值的方法
func BulkQueryUserTagValue(tagNames []string, userId int64, queryArgs ...interface{}) (tagValuePairs []*TagValuePair) {
    tagCount := len(tagNames)
    if tagCount < 1 {
        return
    }
    wg := &sync.WaitGroup{}
    wg.Add(tagCount)
    tagValueCh := make(chan *TagValuePair, tagCount) // 用于接收所有Picker查到的标签值的Channel
    ctx, _ := context.WithTimeout(context.Background(), time.Minute) // 设置执行标签值查找的超时时间
    for _, tagName := range tagNames {
        go queryTagValue(ctx, wg, tagName, userId, tagValueCh, queryArgs...)
    }
    wg.Wait()
    close(tagValueCh) // 先关闭通道 方便下面for range不发生阻塞, 从channel中读完值即退出
    tagValuePairs = make([]*TagValuePair, 0)
    for tagValue := range tagValueCh {
        if tagValue.Value != nil {
            tagValuePairs = append(tagValuePairs, tagValue)
        }
    }

    return tagValuePairs
}

type TagValuePair struct {
    Name string `json:"tag_name"`
    Value interface{} `json:"tag_value"`
}

func queryTagValue(ctx context.Context, wg *sync.WaitGroup, tagName string, userId int64, tagValueCh chan *TagValuePair, queryArgs ...interface{}) {
    defer wg.Done()
    tagPicker := resolveTagPicker(tagName)
    if tagPicker == nil {
        dlog.Error("未识别的业务标签", common.ErrUnknownBusinessTag)
        return
    }
    go tagPicker.PickTagValueForUser(userId, queryArgs...)
    select {
    case <- ctx.Done(): // 超时返回
        return
    case tagValue := <- tagPicker.Notify(): // 接收标签值
        TagValuePair := &TagValuePair{
            Name:  tagName,
            Value: tagValue,
        }
        tagValueCh <- TagValuePair

        return
    }
}

最后就是具体标签查询器的实现了，每个标签都有自己的实现逻辑，下面是OrderNumTagPicker的示例代码：

type OrderNumTagPicker struct {
    TagName string
    ValueCh chan interface{}
}

type TradeNoInfo struct {
    TradeNo string `json:"trade_no"`
}

func (picker *OrderNumTagPicker) PickTagValueForUser(userId int64, args ...interface{}) {
    // 用类型转换得到交易号
        // tradeNo, ok := args[0].(string)
        extInfoJson, _ := args[0].(string)

    if err := json.Unmarshal([]byte(extInfoJson), &TradeNoInfo); err != nil {
                // log.Error自己实现 
        log.Error("PayTotalTagPickerError", "Invalid arg", args[0])
        // 结束执行并通知外部
        picker.ValueCh <- nil
        return
    }
        // 这里就打印下参数值，标签查询的具体逻辑自己实现
        fmt.Println(userId)
        fmt.Println(TradeNo)
    // 查询到的用户的标签 (假设交易号下有10个订单）
    picker.ValueCh <- 10
}

func (picker *OrderNumTagPicker) Notify () <-chan interface{} {
    return picker.ValueCh
}

完整的源代码参考： https://github.com/kevinyan815/gocookbook/tree/master/codes/tag_picker