How to solve the problem of dynamic expansion of concurrent tasks in Go language?

How to solve the problem of dynamic expansion of concurrent tasks in Go language?

When we need to handle a large number of concurrent tasks, we may need to dynamically adjust the number of concurrent goroutines to achieve efficient processing of tasks. In the Go language, you can use goroutine and channel to implement concurrent programming. By adjusting the number of goroutines, you can effectively control the execution of concurrent tasks.

In order to solve the problem of dynamic expansion of concurrent tasks, we can use a goroutine pool to manage the number of concurrent goroutines, and use channels to distribute tasks and collect results. The following is a sample code:

package main

import (
    "fmt"
    "sync"
    "time"
)

type Pool struct {
    queue chan Job
    wg    sync.WaitGroup
}

type Job struct {
    id     int
    result string
}

func NewPool(maxWorkers int) *Pool {
    pool := &Pool{
        queue: make(chan Job),
    }

    for i := 0; i < maxWorkers; i++ {
        go pool.worker(i)
    }

    return pool
}

func (p *Pool) worker(id int) {
    for job := range p.queue {
        fmt.Printf("Worker %d processing job %d
", id, job.id)
        time.Sleep(time.Second) // 模拟任务耗时
        job.result = fmt.Sprintf("Job %d processed by worker %d", job.id, id)
        p.wg.Done()
    }
}

func (p *Pool) AddJob(job Job) {
    p.wg.Add(1)
    p.queue <- job
}

func (p *Pool) Wait() {
    p.wg.Wait()
    close(p.queue)
}

func main() {
    pool := NewPool(3)

    for i := 1; i <= 10; i++ {
        job := Job{id: i}
        pool.AddJob(job)
    }

    pool.Wait()
}

In the above sample code, we defined a Pool structure to manage the goroutine pool, which contains a channel for storing tasks and a user sync.WaitGroup that waits for all tasks to complete.

NewPool function is used to create a new goroutine pool, which will create a corresponding number of goroutines based on the specified maxWorkers parameters and call workerFunction performs task processing. The

worker function is the main function of each goroutine. It obtains tasks from the task channel and processes the tasks. Before processing the task, some preprocessing or other operations can be performed according to specific needs. After the task processing is completed, assign the result to the job.result field, and notify the task completion through the Done method of sync.WaitGroup. The

AddJob method is used to add new tasks to the task channel. It will add waiting tasks through the Add method of sync.WaitGroup quantity and put the task into the queue. The

Wait method is used to wait for all tasks to be completed. It will call the Wait method of sync.WaitGroup to block the main thread until all tasks are completed. All are completed.

Finally, in the main function, we create a goroutine pool of size 3 and add 10 tasks. By adjusting the value of the maxWorkers parameter, we can dynamically adjust the number of concurrent goroutines.

Through the above example code, we can easily solve the problem of dynamic expansion of concurrent tasks. By reasonably controlling the number of concurrent goroutines, we can use the concurrency mechanism of the Go language to achieve efficient task processing.

The above is the detailed content of How to solve the problem of dynamic expansion of concurrent tasks in Go language?. For more information, please follow other related articles on the PHP Chinese website!