程序因频道而挂起

大家好，我们又见面了啊~本文《程序因频道而挂起》的内容中将会涉及到等等。如果你正在学习Golang相关知识，欢迎关注我，以后会给大家带来更多Golang相关文章，希望我们能一起进步！下面就开始本文的正式内容~

我想使用 goroutine 来批量处理来自不同客户的不同日期的请求。

我的意思是 50 个消费者 goroutine 来消费数据库中的所有客户，以及 2 个日期消费者 goroutine 来消费日期切片。

主要代码如下，但它挂起并且没有按预期退出。

为什么没有按预期退出？

func Run(){
    var syncWg sync.WaitGroup
    syncWg.Add(1)
    go SyncCustomerMetricsHistory(&syncWg)
    syncWg.Wait()
}

func SyncCustomerMetricsHistory(wg *sync.WaitGroup){
    defer wg.Done()
    odb := orm.NewOrm()
    start := time.Now()
    logs.Info("start sync  customer metrics, time:[%v]", start)

    qs := odb.QueryTable("gg_customer")
    var customers []*db.GgCustomer
    if num, err := qs.All(&customers); err != nil || num == 0 {
        logs.Error("Get customer error, rows:[%v], err:[%v]", num, err)
    }

    customersChan := make(chan *db.GgCustomer, 50)

    var wgC sync.WaitGroup
    wgC.Add(50)
    for i := 0; i < 50; i++ {
        go syncCustomerMetricsHistory(customersChan, &wgC)
    }

    go func() {
        for _, customer := range customers {
            customersChan <- customer
        }
        close(customersChan)
    }()

    wgC.Wait()
}

func  syncCustomerMetricsHistory(customerChan <- chan *db.GgCustomer, wg *sync.WaitGroup){
    defer wg.Done()
    for customer := range customerChan{
            dateChan := make(chan string, 2)
            var wgD sync.WaitGroup
            wgD.Add(2)
            for i := 1; i < 2; i++{
                go test(dateChan, customer, &wgD)
            }
            go func(){
                for _, date := range GetAllYearDate(){
                    dateChan <- date
                }
                close(dateChan)
            }()

            wgD.Wait()
        }
    }
}

func test(dateChan <- chan string, customer *db.GgCustomer, wg *sync.WaitGroup){
    defer wg.Done()
    for date := range dateChan{
        fmt.Println(date, customer)
    }
}


func  GetAllYearDate()  []string{
  return []string{"2019-10-01", "2019-10-02"}
}

解决方案

我还没有尝试运行它（因为它需要额外的代码），但相信您的问题是：

wgd.add(2)
for i := 1; i < 2; i++{
 go test(datechan, customer, &wgd)
}

那个 for 循环只会迭代一次，但你调用了 wgd.add(2) （我认为你可能意味着循环迭代两次；尝试 i <= 2）。

另一点反馈；您使用等待组的方式会起作用，但很难遵循（可能导致您没有发现问题）；怎么样：

func Run(){
    SyncCustomerMetricsHistory()  // No wait group needed as this will not return before done
}

func SyncCustomerMetricsHistory(){
    odb := orm.NewOrm()
    start := time.Now()
    logs.Info("start sync  customer metrics, time:[%v]", start)

    qs := odb.QueryTable("gg_customer")
    var customers []*db.GgCustomer
    if num, err := qs.All(&customers); err != nil || num == 0 {
        logs.Error("Get customer error, rows:[%v], err:[%v]", num, err)
    }

    customersChan := make(chan *db.GgCustomer, 50)

    var wgC sync.WaitGroup
    wgC.Add(50)
    for i := 0; i < 50; i++ {
        go func() {
            syncCustomerMetricsHistory(customersChan)
            wgC.Done()
        }()
    }

    go func() {
        for _, customer := range customers {
            customersChan <- customer
        }
        close(customersChan)
    }()
    wgC.Wait()
}



func  syncCustomerMetricsHistory(customerChan <- chan *db.GgCustomer){
    for customer := range customerChan{
            dateChan := make(chan string, 2)
            var wgD sync.WaitGroup
            wgD.Add(2)
            for i := 1; i < 2; i++{
                go func() {
                    test(dateChan, customer)
                    wgD.Done()
                }()
            }
            go func(){
                for _, date := range GetAllYearDate(){
                    dateChan <- date
                }
                close(dateChan)
            }()
            wgD.Wait()
        }
    }
}

我认为这更容易理解，因为您可以看到 wg.done() 被调用的位置。在两侧粘贴一些 fmt.println 命令也非常容易，这使得调试此类问题变得更简单。

以上就是本文的全部内容了，是否有顺利帮助你解决问题？若是能给你带来学习上的帮助，请大家多多支持golang学习网！更多关于Golang的相关知识，也可关注golang学习网公众号。