通过Go channel批量读取数据的示例详解

package main

import (
	"fmt"
	"time"
)

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

	for {
		select {
		case data := <-ch:
			batch = append(batch, data)
			// 当缓冲区达到批量大小时处理
			if len(batch) >= batchSize {
				fmt.Printf("Processing batch: %v\n", batch)
				batch = nil
				// 重置定时器
				timer.Reset(flushInterval)
			}

		case <-timer.C:
			// 如果达到时间间隔，但 batch 不为空，也进行处理
			if len(batch) > 0 {
				fmt.Printf("Processing batch on timer: %v\n", batch)
				batch = nil
			}
			// 重置定时器
			timer.Reset(flushInterval)
		}
	}
}

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

	// 启动批量处理协程
	go batchProcessor(dataChannel, batchSize, flushInterval)

	// 模拟向 channel 发送数据
	for i := 1; i <= 10; i++ {
		dataChannel <- fmt.Sprintf("data-%d", i)
		time.Sleep(1 * time.Second)
	}

	// 让主程序暂停一会，以便查看处理结果
	time.Sleep(5 * time.Second)
}

go get github.com/Shopify/sarama

package main

import (
	"fmt"
	"log"
	"time"

	"github.com/Shopify/sarama"
)

// 初始化 Kafka 生产者
func initKafkaProducer(brokers []string) sarama.SyncProducer {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	producer, err := sarama.NewSyncProducer(brokers, config)
	if err != nil {
		log.Fatalf("Failed to start Kafka producer: %v", err)
	}
	return producer
}

// 批量发送消息到 Kafka
func sendBatchToKafka(producer sarama.SyncProducer, topic string, messages []string) {
	var kafkaMessages []*sarama.ProducerMessage
	for _, msg := range messages {
		kafkaMessages = append(kafkaMessages, &sarama.ProducerMessage{
			Topic: topic,
			Value: sarama.StringEncoder(msg),
		})
	}

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

// 批量处理 Kafka 消息
func kafkaBatchProcessor(producer sarama.SyncProducer, topic string, ch <-chan string, batchSize int, flushInterval time.Duration) {
	var batch []string
	timer := time.NewTimer(flushInterval)

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

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

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

	// 初始化 Kafka 生产者
	producer := initKafkaProducer(brokers)
	defer producer.Close()

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

	// 启动 Kafka 批量处理协程
	go kafkaBatchProcessor(producer, topic, dataChannel, batchSize, flushInterval)

	// 模拟向 channel 发送数据
	for i := 1; i <= 10; i++ {
		dataChannel <- fmt.Sprintf("message-%d", i)
		time.Sleep(1 * time.Second)
	}

	// 让主程序暂停一会以便查看处理结果
	time.Sleep(5 * time.Second)
}

package main

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

// 批量发送 HTTP 请求
func sendBatchToAPI(endpoint string, batch []string) {
	// 构造请求体
	var requestBody bytes.Buffer
	for _, data := range batch {
		requestBody.WriteString(fmt.Sprintf("%s\n", data))
	}

	// 发送 HTTP POST 请求
	resp, err := http.Post(endpoint, "text/plain", &requestBody)
	if err != nil {
		log.Printf("Failed to send batch: %v", err)
		return
	}
	defer resp.Body.Close()

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

// 批量处理 HTTP 请求
func httpBatchProcessor(endpoint string, ch <-chan string, batchSize int, flushInterval time.Duration) {
	var batch []string
	timer := time.NewTimer(flushInterval)

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

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

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

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

	// 启动 HTTP 批量处理协程
	go httpBatchProcessor(apiEndpoint, dataChannel, batchSize, flushInterval)

	// 模拟向 channel 发送数据
	for i := 1; i <= 10; i++ {
		dataChannel <- fmt.Sprintf("data-%d", i)
		time.Sleep(1 * time.Second)
	}

	// 让主程序暂停一会以便查看处理结果
	time.Sleep(5 * time.Second)
}