Всегда иметь х число запущенных в любое время

Ответ 1

Спасибо всем за то, что помогли мне в этом. Тем не менее, я не чувствую, что кто-то действительно предоставил что-то, что сработало и было простым/понятным, хотя вы все помогли мне понять технику.

То, что я сделал в конце, я считаю гораздо более понятным и практичным в качестве ответа на мой конкретный вопрос, поэтому я отправлю его здесь, если у кого-то будет тот же вопрос.

Каким-то образом это оказалось очень похоже на то, что OneOfOne отправил, что здорово, потому что теперь я это понимаю. Но код OneOfOne, который мне сначала очень трудно понять из-за передачи функций функциям, затрудняет понимание того, для чего был бит. Я думаю, что этот способ имеет больший смысл:

package main

import (
"fmt"
"sync"
)

const xthreads = 5 // Total number of threads to use, excluding the main() thread

func doSomething(a int) {
    fmt.Println("My job is",a)
    return
}

func main() {
    var ch = make(chan int, 50) // This number 50 can be anything as long as it larger than xthreads
    var wg sync.WaitGroup

    // This starts xthreads number of goroutines that wait for something to do
    wg.Add(xthreads)
    for i:=0; i<xthreads; i++ {
        go func() {
            for {
                a, ok := <-ch
                if !ok { // if there is nothing to do and the channel has been closed then end the goroutine
                    wg.Done()
                    return
                }
                doSomething(a) // do the thing
            }
        }()
    }

    // Now the jobs can be added to the channel, which is used as a queue
    for i:=0; i<50; i++ {
        ch <- i // add i to the queue
    }

    close(ch) // This tells the goroutines there nothing else to do
    wg.Wait() // Wait for the threads to finish
}

Ответ 2

Вы можете найти Перейти Concurrency Шаблоны в статье, интересной, особенно ограниченной parallelism, она объясняет точный шаблон, который вам нужен.

Вы можете использовать канал пустых структур как ограничивающий страж для контроля количества одновременных рабочих горок:

package main

import "fmt"

func main() {
    maxGoroutines := 10
    guard := make(chan struct{}, maxGoroutines)

    for i := 0; i < 30; i++ {
        guard <- struct{}{} // would block if guard channel is already filled
        go func(n int) {
            worker(n)
            <-guard
        }(i)
    }
}

func worker(i int) { fmt.Println("doing work on", i) }

Ответ 3

• Создайте канал для передачи данных в goroutines.
• Запустите 20 goroutines, которые обрабатывают данные из канала в цикле.
• Отправляйте данные на канал вместо запуска новой версии goroutine.

Ответ 4

Здесь я думаю, что что-то простое, как это будет работать:

package main

import "fmt"

const MAX = 20

func main() {
    sem := make(chan int, MAX)
    for {
        sem <- 1 // will block if there is MAX ints in sem
        go func() {
            fmt.Println("hello again, world")
            <-sem // removes an int from sem, allowing another to proceed
        }()
    }
}

Ответ 5

Grzegorz Żur answer - наиболее эффективный способ сделать это, но для новичка это может быть трудно реализовать без чтения кода, поэтому здесь очень простая реализация:

type idProcessor func(id uint)

func SpawnStuff(limit uint, proc idProcessor) chan<- uint {
    ch := make(chan uint)
    for i := uint(0); i < limit; i++ {
        go func() {
            for {
                id, ok := <-ch
                if !ok {
                    return
                }
                proc(id)
            }
        }()
    }
    return ch
}

func main() {
    runtime.GOMAXPROCS(4)
    var wg sync.WaitGroup //this is just for the demo, otherwise main will return
    fn := func(id uint) {
        fmt.Println(id)
        wg.Done()
    }
    wg.Add(1000)
    ch := SpawnStuff(10, fn)
    for i := uint(0); i < 1000; i++ {
        ch <- i
    }
    close(ch) //should do this to make all the goroutines exit gracefully
    wg.Wait()
}

playground

Ответ 6

Это простая проблема производитель-потребитель, которая в Go может быть легко решена с использованием каналов для буферизации пакетов.

Проще говоря: создайте канал, который принимает ваши идентификаторы. Запустите несколько подпрограмм, которые будут считываться из канала в цикле, затем обработать ID. Затем запустите свой цикл, который будет передавать идентификаторы на канал.

Пример:

func producer() {
    var buffer = make(chan uint)

    for i := 0; i < 20; i++ {
        go consumer(buffer)
    }

    for _, id :=  range IDs {
        buffer <- id
    }
}

func consumer(buffer chan uint) {
    for {
        id := <- buffer
        // Do your things here
    }
}

Что нужно знать:

• Небуферизованные каналы блокируют: если элемент, записанный в канал, не принимается, процедура подачи элемента блокируется до тех пор, пока не будет
• В моем примере отсутствует механизм закрытия: вы должны найти способ заставить продюсера ждать, пока все потребители закончат цикл, прежде чем вернуться. Самый простой способ сделать это - с другим каналом. Я позволяю вам думать об этом.

Ответ 7

Я написал простой пакет для обработки параллелизма для Голанга. Этот пакет поможет вам ограничить количество подпрограмм, которые могут запускаться одновременно: https://github.com/zenthangplus/goccm

Пример:

package main

import (
    "fmt"
    "goccm"
    "time"
)

func main()  {
    // Limit 3 goroutines to run concurrently.
    c := goccm.New(3)

    for i := 1; i <= 10; i++ {

        // This function have to call before any goroutine
        c.Wait()

        go func(i int) {
            fmt.Printf("Job %d is running\n", i)
            time.Sleep(2 * time.Second)

            // This function have to when a goroutine has finished
            // Or you can use 'defer c.Done()' at the top of goroutine.
            c.Done()
        }(i)
    }

    // This function have to call to ensure all goroutines have finished 
    // after close the main program.
    c.WaitAllDone()
}