使用通道来同步多个 goroutine

从现在开始，努力学习吧！本文《使用通道来同步多个 goroutine》主要讲解了等等相关知识点，我会在golang学习网中持续更新相关的系列文章，欢迎大家关注并积极留言建议。下面就先一起来看一下本篇正文内容吧，希望能帮到你！

我有以下用例：

1. “fetch”goroutine 将根据一些预定义的标准从数据库中获取可用数据。
2. 然后我将拥有 2 个 goroutine（process1、process2），每个 goroutine 都会对数据进行一些操作（并且顺序很重要）。
3. 最后一个 goroutine (processsave) 应该更新数据库中的数据。

我知道我需要使用原始通道将每个 goroutine 与下一个 goroutine 连接起来： fetchtop1chnl、p1top2chnl、p2top3chnl、p3tosavechnl。

对“工作”对象的操作必须以顺序方式运行： fetch -> process1 -> process2 -> processsave 。

更新： 我忘了提及 process1 和 process2 函数正在更改对象。 process2 需要处理 process1 的结果。

我不确定的问题：

• 这里更适合哪种通道：无缓冲通道或缓冲通道（是缓冲通道，那么如何选择最佳大小）
• 在哪里更好地打开这些渠道？ （我认为应该在 main 中完成）
• 哪里最好放置通道的关闭位置？我的应用程序预计会不间断运行

我正在考虑以下方法：

type Object struct {
    ID           string `bson:"_id"`
    Data         string `bson:"data"`
    Subdocument1 string // Added by process1
    Subdocument2 string // Added by process2
}

func main() {
    // Set up MongoDB client
    clientOptions := options.Client().ApplyURI("mongodb://localhost:27017")
    client, err := mongo.Connect(context.Background(), clientOptions)
    if err != nil {
        log.Fatal(err)
    }

    // Fetch objects from MongoDB
    collection := client.Database("your-database").Collection("your-collection")
    cursor, err := collection.Find(context.Background(), bson.M{})
    if err != nil {
        log.Fatal(err)
    }

    // Create channels for communication between goroutines
    objectCh1 := make(chan Object)
    objectCh2 := make(chan Object)

    // Create wait group to wait for goroutines to finish
    var wg sync.WaitGroup

    // Start the fetcher goroutine
    wg.Add(1)
    go fetchObjects(cursor, objectCh1, &wg)

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

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

    // Wait for all goroutines to finish
    wg.Wait()

    // Close MongoDB client
    if err := client.Disconnect(context.Background()); err != nil {
        log.Fatal(err)
    }

    fmt.Println("Done")
}

// fetchObjects fetches objects from the cursor and sends them to 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"
        // Do additional processing or update in 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"
        // Do additional processing or update in MongoDB
        updateObjectInMongoDB(obj)
    }
}

// updateObjectInMongoDB is a placeholder function to update the object in MongoDB
func updateObjectInMongoDB(obj Object) Object {
    fmt.Printf("Updated Object: %+v\n", obj)
    // Your logic to update the object in MongoDB
}

正确答案

通道：您可能需要缓冲通道，以便即使读取器暂时繁忙，发送者 goroutine 也可以继续工作（我假设您想增加并发性）。

在哪里打开通道：在 goroutine 本身之外“连接” goroutine/通道图是一种很好的做法，所以是的，可能在 main 中。

关闭通道：如果您的应用程序不停地运行，那么您可能根本不需要关闭通道。但这很棘手——大多数服务应用程序并不会真正永远运行。您可能想要实现正常关闭（例如响应信号）。经验法则是发送者有责任关闭通道。

还有一点：为什么需要以这种方式设计并发性并不明显。您可以拥有一个 fetcher goroutine（我假设数据库读取不应并行化），然后将对象拆分为多个工作 goroutine，每个 goroutine 执行 process1、process2 并在给定对象上按顺序保存。

终于介绍完啦！小伙伴们，这篇关于《使用通道来同步多个 goroutine》的介绍应该让你收获多多了吧！欢迎大家收藏或分享给更多需要学习的朋友吧~golang学习网公众号也会发布Golang相关知识，快来关注吧！