进入频道非阻塞多个接收

Question

Everywhere seems to discuss that reading from a channel should always be a blocking operation. The attitude seems to be this is the Go way. This makes some sense but I'm trying to figure out how I would aggregate things from channels.

For example, sending http requests. Say I have a pipeline setup that generates streams of data, so I have a channel that produces queue/stream of points. I could then have a goroutine listen to this channel and send a HTTP Request to store it in a service. This works, but I'm creating a http request for every point.

The endpoint I'm sending it too allows me to send multiple data points in a batch. What I would like to do, is

1. Read as many values until I would block on channel.
2. Combine them/send single http request.
3. Then block on channel until I can read one again.

This is how I would've done things in C, with threadsafe queues and select statements. Basically flushing the entire/queue buffer when possible. Is this a valid technique in go?

It seems the go select statement does give me something similar to C's select, but I'm still not sure if there is a 'nonblocking read' on channels.

EDIT: I'm also willing to accept what I'm intending may not be the Go Way, but constantly smashing off non stop http requests also seems wrong to me, especially if they can be aggregated. If someone has an alternative architecture that will be cool, but I want to avoid things like, magically buffering N items, or waiting X seconds until sending.

Answer 1

Here's how to batch until the channel is empty. The variable batch is a slice of your data point type. The variable ch is a channel of your data point type.

var batch []someType
for {
    select {
    case v := <-ch:
       batch = append(batch, v)
    default:
       if len(batch) > 0 {
           sendBatch(batch)
           batch := batch[:0]
       }
       batch = append(batch, <-ch)  // receiving a value here prevents busy waiting.
    }
}

You should prevent the batch from growing without limit. Here's a simple way to do it:

var batch []someType
for {
    select {
    case v := <-ch:
       batch = append(batch, v)
       if len(batch) >= batchLimit {
           sendBatch(batch)
           batch := batch[:0]
       }
    default:
       if len(batch) > 0 {
           sendBatch(batch)
           batch := batch[:0]
       }
       batch = append(batch, <-ch)
    }
}