Выход канала дозирования в Go - PullRequest
Новая
       13

Выход канала дозирования в Go

0 голосов
/ 03 февраля 2020

У меня есть подпрограмма, которая потребляет элементы канала. Элементы доступны в произвольное время (канал подключен к сетевому сокету, где разные клиенты обеспечивают ввод в разное время), и их необходимо проталкивать через API с ограниченной скоростью, который может принимать партии элементов одновременно.

Мое текущее решение состоит в том, чтобы использовать контейнер списка и тикер: goroutine извлекает элементы из канала и выталкивает их в список, затем тикер, который запускает каждое «минимальное количество времени, чтобы не удушаться», захватывает содержимое перечислите его и добавьте в API с ограниченной скоростью.

Я придумал это (извиняюсь, я не программист Go): 

// simulate the channel
c := make(chan int, 100)
go func() {
    for i := 1; i < 30000; i++ {
        c <- i
        time.Sleep(1 * time.Millisecond)
    }
}()

// block-read the channel
l := []int{}
var m sync.Mutex
go func(list *[]int) {
    for val := range c {
        m.Lock()
        *list = append(*list, val)
        m.Unlock()
    }
}(l)

// read batches every tick
for range time.Tick(1 * time.Second) {
    buf := append([]int{}, l...)
    m.Lock()
    l = l[len(buf):]
    m.Unlock()
    sendBuffer(buf)
}

Это нормально? ?

Я также думал о том, чтобы использовать select неблокирующий опрос, чтобы «прочитать все доступные сообщения», а затем отправить и уснуть, но описанный выше способ показался более go -i sh, может быть?

Кроме того, какие очевидные ловушки я упускаю из-за моего непонимания Go?

1 Ответ

4 голосов
/ 03 февраля 2020

Вы можете сделать что-то подобное 

c := make(chan int, 100)
go func() {
    for i := 1; i < 30000; i++ {
        c <- i
        time.Sleep(1 * time.Millisecond)
    }
}()

ticker := time.NewTicker(time.Second)
defer ticker.Stop() // release resources

data := make([]int, 0)

for {
    select {
        case <- ticker.C:
            sendBuffer(data)
            data = make([]int, 0)
        case i := <- c:
            data = append(data, i)
    }
}

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

...