当前位置：首页 > 文章列表 > Golang > Go问答 > 使用通道来同步多个 goroutine

使用通道来同步多个 goroutine

来源：stackoverflow 2024-02-28 22:21:23 0浏览收藏

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

问题内容

我有以下用例：

“fetch”goroutine 将根据一些预定义的标准从数据库中获取可用数据。
然后我将拥有 2 个 goroutine（process1、process2），每个 goroutine 都会对数据进行一些操作（并且顺序很重要）。
最后一个 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
}