SoFunction
Updated on 2025-03-03

How to read data in batches of Go channel

In Go, we can usechannelAs a transmission channel of data, it is read in batches regularlychanneland send the data in batches to Kafka or write to the network. This can improve system performance and reduce network overhead for a single request.

The main logic of batch processing is:channelAfter receiving data in the data, accumulate to a certain amount or reaching the time limit, the data will be processed in batches (for example, sending it to Kafka or writing it to the network).

I'll show one fromGo channelExamples of batch reading of data and batch sending to Kafka and batch writing to network data.

1. General logic for batch reading of Go channel

Batch readingGo channelThe general logic can be implemented by a timer and a buffer:

  • When the number of buffers reaches a predetermined value, batch operations are performed.
  • When the time exceeds a certain predetermined time interval, batch processing is performed even if the buffer is not full.
package main

import (
	"fmt"
	"time"
)

func batchProcessor(ch <-chan string, batchSize int, flushInterval ) {
	var batch []string
	timer := (flushInterval)

	for {
		select {
		case data := <-ch:
			batch = append(batch, data)
			// Process when the buffer reaches batch size			if len(batch) >= batchSize {
				("Processing batch: %v\n", batch)
				batch = nil
				// Reset the timer				(flushInterval)
			}

		case <-:
			// If the time interval is reached, but batch is not empty, it will also be processed			if len(batch) > 0 {
				("Processing batch on timer: %v\n", batch)
				batch = nil
			}
			// Reset the timer			(flushInterval)
		}
	}
}

func main() {
	dataChannel := make(chan string)
	batchSize := 5
	flushInterval := 3 * 

	// Start the batch processing coroutine	go batchProcessor(dataChannel, batchSize, flushInterval)

	// Simulate sending data to channel	for i := 1; i <= 10; i++ {
		dataChannel <- ("data-%d", i)
		(1 * )
	}

	// Let the main program pause for a while to view the processing results	(5 * )
}

The above code showschannelThe basic mechanism for batch reading of data:

  • Buffer size: Batch processing is triggered when the buffer is full.
  • Time interval: When the specified time interval is reached, batch processing is triggered even if the buffer is not full.

2. Send data in batches to Kafka

Based on batch processing logic, we can use the Kafka client library to implement batch sending messages to Kafka.

use/Shopify/saramaIt is a commonly used Kafka client library in Go. Install it first:

go get /Shopify/sarama

Then implement an example of batch sending data to Kafka:

package main

import (
	"fmt"
	"log"
	"time"

	"/Shopify/sarama"
)

// Initialize Kafka Producerfunc initKafkaProducer(brokers []string)  {
	config := ()
	 = true
	producer, err := (brokers, config)
	if err != nil {
		("Failed to start Kafka producer: %v", err)
	}
	return producer
}

// Send messages in batches to Kafkafunc sendBatchToKafka(producer , topic string, messages []string) {
	var kafkaMessages []*
	for _, msg := range messages {
		kafkaMessages = append(kafkaMessages, &{
			Topic: topic,
			Value: (msg),
		})
	}

	err := (kafkaMessages)
	if err != nil {
		("Failed to send messages: %v", err)
	} else {
		("Successfully sent batch to Kafka: %v", messages)
	}
}

// Batch processing of Kafka messagesfunc kafkaBatchProcessor(producer , topic string, ch <-chan string, batchSize int, flushInterval ) {
	var batch []string
	timer := (flushInterval)

	for {
		select {
		case msg := <-ch:
			batch = append(batch, msg)
			if len(batch) >= batchSize {
				sendBatchToKafka(producer, topic, batch)
				batch = nil
				(flushInterval)
			}

		case <-:
			if len(batch) > 0 {
				sendBatchToKafka(producer, topic, batch)
				batch = nil
			}
			(flushInterval)
		}
	}
}

func main() {
	// Kafka broker and topic configuration	brokers := []string{"localhost:9092"}
	topic := "test_topic"

	// Initialize Kafka Producer	producer := initKafkaProducer(brokers)
	defer ()

	dataChannel := make(chan string)
	batchSize := 5
	flushInterval := 3 * 

	// Start Kafka batch processing coroutine	go kafkaBatchProcessor(producer, topic, dataChannel, batchSize, flushInterval)

	// Simulate sending data to channel	for i := 1; i <= 10; i++ {
		dataChannel <- ("message-%d", i)
		(1 * )
	}

	// Let the main program pause for a while to view the processing results	(5 * )
}

In this example:

  • kafkaBatchProcessorFunction batch fromchannelRead data in and send messages to Kafka when the batch size reaches or the time interval arrives.
  • Usedto ensure that messages are sent successfully in batches.

3. Batch writing of network data

The same logic can be used to write network data in batches. For example, write data in batches to a certain HTTP API.

Here we use Gonet/httpTo implement batch sending HTTP requests:

package main

import (
	"bytes"
	"fmt"
	"log"
	"net/http"
	"time"
)

// Batch send HTTP requestsfunc sendBatchToAPI(endpoint string, batch []string) {
	// Construct the request body	var requestBody 
	for _, data := range batch {
		(("%s\n", data))
	}

	// Send HTTP POST request	resp, err := (endpoint, "text/plain", &requestBody)
	if err != nil {
		("Failed to send batch: %v", err)
		return
	}
	defer ()

	("Successfully sent batch to API: %v", batch)
}

// Batch processing of HTTP requestsfunc httpBatchProcessor(endpoint string, ch <-chan string, batchSize int, flushInterval ) {
	var batch []string
	timer := (flushInterval)

	for {
		select {
		case msg := <-ch:
			batch = append(batch, msg)
			if len(batch) >= batchSize {
				sendBatchToAPI(endpoint, batch)
				batch = nil
				(flushInterval)
			}

		case <-:
			if len(batch) > 0 {
				sendBatchToAPI(endpoint, batch)
				batch = nil
			}
			(flushInterval)
		}
	}
}

func main() {
	// API endpoint
	apiEndpoint := "http://localhost:8080/receive"

	dataChannel := make(chan string)
	batchSize := 5
	flushInterval := 3 * 

	// Start HTTP batch processing coroutine	go httpBatchProcessor(apiEndpoint, dataChannel, batchSize, flushInterval)

	// Simulate sending data to channel	for i := 1; i <= 10; i++ {
		dataChannel <- ("data-%d", i)
		(1 * )
	}

	// Let the main program pause for a while to view the processing results	(5 * )
}

Summarize

The above shows the implementation of batch reading of data through Go channel and batch sending it to Kafka or HTTP API:

  • Process data in batchesIt can significantly reduce frequent network requests and improve performance.
  • useTimerTo ensure that data can be sent out on time even if the batch size is not reached.

This architecture is very suitable for high-throughput task processing scenarios, such as logging systems, data processing pipelines, etc.

This is the end of this article about how Go channel reads data in batches. For more information about Go channel reading data in batches, please search for my previous articles or continue browsing the related articles below. I hope everyone will support me in the future!