基于Channels组合多个Goroutines的实现方案与疑问

业务场景 #

• "fetch" goroutine 根据预设条件从数据库获取可用数据
• 随后启动2个goroutine（process1、process2），各自对数据进行处理，处理顺序至关重要，必须严格遵循 fetch → process1 → process2 → processSave 的执行流程
• 最后由processSave goroutine负责将处理完成的数据更新回数据库

补充说明：process1和process2会修改目标对象，process2必须基于process1处理后的结果进行操作

核心疑问 #

• 哪种类型的通道更适合此场景：无缓冲通道还是有缓冲通道？若选择有缓冲通道，如何确定最优大小？
• 通道最好在何处创建？（我认为应该在main函数中创建）
• 通道最好在何处关闭？我的应用需要持续运行

我的实现构思 #

type Object struct {
    ID           string `bson:"_id"`
    Data         string `bson:"data"`
    Subdocument1 string // 由process1添加
    Subdocument2 string // 由process2添加
}

func main() {
    // 初始化MongoDB客户端
    clientOptions := options.Client().ApplyURI("mongodb://localhost:27017")
    client, err := mongo.Connect(context.Background(), clientOptions)
    if err != nil {
        log.Fatal(err)
    }

    // 从MongoDB查询数据
    collection := client.Database("your-database").Collection("your-collection")
    cursor, err := collection.Find(context.Background(), bson.M{})
    if err != nil {
        log.Fatal(err)
    }

    // 创建goroutine间通信的通道
    objectCh1 := make(chan Object)
    objectCh2 := make(chan Object)

    // 初始化等待组，用于等待所有goroutine完成
    var wg sync.WaitGroup

    // 启动fetch goroutine
    wg.Add(1)
    go fetchObjects(cursor, objectCh1, &wg)

    // 启动process1 goroutine
    wg.Add(1)
    go process1(objectCh1, objectCh2, &wg)

    // 启动process2 goroutine
    wg.Add(1)
    go process2(objectCh2, &wg)

    // 等待所有goroutine执行完毕
    wg.Wait()

    // 关闭MongoDB客户端连接
    if err := client.Disconnect(context.Background()); err != nil {
        log.Fatal(err)
    }

    fmt.Println("处理完成")
}

// fetchObjects 从游标中读取数据并发送到objectCh1
func fetchObjects(cursor *mongo.Cursor, objectCh1 chan<- Object, wg *sync.WaitGroup) {
    defer close(objectCh1)
    defer wg.Done()

    for cursor.Next(context.Background()) {
        var obj Object
        err := cursor.Decode(&obj)
        if err != nil {
            log.Println("解码数据失败:", err)
            continue
        }

        objectCh1 <- obj
    }

    if err := cursor.Err(); err != nil {
        log.Println("游标遍历出错:", err)
    }
}

func process1(objectCh1 <-chan Object, objectCh2 chan<- Object, wg *sync.WaitGroup) {
    defer wg.Done()

    for obj := range objectCh1 {
        obj.Subdocument1 = "subdocument1"
        // 执行额外处理或更新MongoDB
        updatedObject := updateObjectInMongoDB(obj)
        objectCh2 <- updatedObject
    }

    close(objectCh2)
}

func process2(objectCh2 <-chan Object, wg *sync.WaitGroup) {
    defer wg.Done()

    for obj := range objectCh2 {
        obj.Subdocument2 = "subdocument2"
        // 执行额外处理或更新MongoDB
        updateObjectInMongoDB(obj)
    }
}

// updateObjectInMongoDB 用于更新MongoDB中对象的占位函数
func updateObjectInMongoDB(obj Object) Object {
    fmt.Printf("已更新对象: %+v\n", obj)
    // 此处编写实际的MongoDB更新逻辑
    return obj
}

疑问解答 #

1. 通道类型选择与缓冲大小 #

• 无缓冲通道：完全匹配你要求的严格顺序执行逻辑——无缓冲通道的发送和接收操作会相互阻塞，确保每个对象必须被前一个步骤处理完成后，下一个步骤才能开始处理。这种方式能保证数据处理顺序绝对一致，不会出现乱序，是你的场景下的首选。
• 有缓冲通道：如果各步骤处理速度差异较大（比如fetch速度远快于process1），可以用有缓冲通道平衡吞吐量，避免快的步骤被慢的步骤阻塞。缓冲大小的选择需结合实际压测调整：
  • 先从**小缓冲（如2-4）**开始测试，观察各goroutine的阻塞情况
  • 如果fetch持续被阻塞，逐步增大缓冲，直到各步骤吞吐量趋于稳定
  • 注意：过大的缓冲会占用更多内存，且可能弱化顺序执行的直观性，但只要通道按顺序传递数据，最终处理顺序依然能保证。

2. 通道创建位置 #

你的判断完全正确，通道应该在main函数（或顶层协调函数）中创建。这样所有goroutine的依赖（通道）都由协调者统一管理，避免goroutine间直接依赖，降低耦合度，也便于后续修改和维护。

3. 通道关闭与持续运行 #

当前实现是一次性处理游标数据，若要让应用持续运行，需调整通道关闭逻辑：

• 避免在fetch中直接关闭通道：持续运行场景下，fetch会不断获取新数据（如定时轮询、监听数据库变更），通道关闭后无法再发送数据，会导致报错。
• 用退出信号控制优雅关闭：新增一个quit通道触发所有goroutine退出。当收到退出信号（如SIGINT）时，先停止fetch的新数据发送，再依次关闭各通道，等待所有goroutine处理完剩余数据后退出。
• 示例调整逻辑：

  func main() {
      quit := make(chan struct{})
      // ... 其他初始化代码
  
      // 启动fetch goroutine，传入quit通道
      wg.Add(1)
      go fetchObjects(collection, objectCh1, quit, &wg)
  
      // 监听退出信号
      sigCh := make(chan os.Signal, 1)
      signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
      go func() {
          <-sigCh
          close(quit) // 触发退出流程
      }()
  
      wg.Wait()
      // ... 关闭MongoDB客户端
  }
  
  func fetchObjects(collection *mongo.Collection, objectCh1 chan<- Object, quit <-chan struct{}, wg *sync.WaitGroup) {
      defer wg.Done()
      for {
          select {
          case <-quit:
              close(objectCh1) // 退出前关闭通道，通知后续goroutine
              return
          default:
              // 轮询获取新数据或监听变更
              cursor, err := collection.Find(context.Background(), bson.M{"status": "pending"})
              if err != nil {
                  log.Println("查询失败:", err)
                  time.Sleep(5 * time.Second) // 失败后重试
                  continue
              }
              for cursor.Next(context.Background()) {
                  var obj Object
                  if err := cursor.Decode(&obj); err != nil {
                      log.Println("解码失败:", err)
                      continue
                  }
                  select {
                  case objectCh1 <- obj:
                  case <-quit:
                      close(objectCh1)
                      cursor.Close(context.Background())
                      return
                  }
              }
              cursor.Close(context.Background())
              time.Sleep(10 * time.Second) // 轮询间隔
          }
      }
  }
  

内容的提问来源于stack exchange，提问作者nikabu