基于Golang协程机制实现高并发场景下的流量统计分析

package main

type TrafficData struct {
	SourceID string
	Count    int64
}

package main

import (
	"fmt"
	"sync"
	"time"
)

// 模拟从数据源读取数据（耗时操作）
func fetchTrafficFromSource(sourceID string) int64 {
	// 模拟网络延迟 200ms
	time.Sleep(200 * time.Millisecond)
	return int64(len(sourceID) * 100) // 模拟随机流量值
}

func main() {
	sources := []string{"Log-A", "Log-B", "Log-C", "Log-D", "Log-E"}
	var totalTraffic int64
	
	var wg sync.WaitGroup
	var mu sync.Mutex // 定义互斥锁

	startTime := time.Now()

	for _, source := range sources {
		wg.Add(1) // 计数器 +1
		go func(id string) {
			defer wg.Done() // 协程结束时计数器 -1

			// 1. 获取数据
			data := fetchTrafficFromSource(id)

			// 2. 写入全局变量（加锁保护）
			mu.Lock()
			totalTraffic += data
			mu.Unlock()
			
			fmt.Printf("Source %s 处理完成\n", id)
		}(source)
	}

	wg.Wait() // 等待所有协程结束

	fmt.Printf("总耗时: %v, 总流量: %d\n", time.Since(startTime), totalTraffic)
}

package main

import (
	"fmt"
	"time"
)

// 生产者：只负责读数据，扔进通道
func producer(id string, ch chan<- TrafficData) {
	data := TrafficData{
		SourceID: id,
		Count:    fetchTrafficFromSource(id),
	}
	ch <- data // 发送数据到通道
	fmt.Printf("Producer: %s 发送数据\n", id)
}

// 消费者：只负责从通道拿数据并汇总
func consumer(ch <-chan TrafficData, done chan<- bool) {
	total := int64(0)
	for data := range ch {
		total += data.Count
		fmt.Printf("Consumer: 收到 %s, 累计流量: %d\n", data.SourceID, total)
	}
	done <- true // 通知主程序汇总完成
}

func main() {
	sources := []string{"Log-A", "Log-B", "Log-C"}
	
	// 创建一个带缓冲的通道，缓冲大小为 5
	// 缓冲通道可以协程解耦，生产者不需要阻塞等待消费者接收
	dataCh := make(chan TrafficData, 5) 
	doneCh := make(chan bool)

	startTime := time.Now()

	// 启动消费者
	go consumer(dataCh, doneCh)

	// 启动生产者
	for _, source := range sources {
		go producer(source, dataCh)
	}

	// 监控：当所有生产者都结束后，关闭通道
	// 注意：实际项目中通常用 WaitGroup 来协调，这里为了简化逻辑
	time.Sleep(1 * time.Second) 
	close(dataCh) // 关闭通道，消费者会结束 for range 循环

	<-doneCh // 等待消费者结束
	fmt.Printf("总耗时: %v\n", time.Since(startTime))
}

package main

import (
	"fmt"
	"time"
)

// 任务：包含任务 ID 和处理逻辑
type Job struct {
	ID   int
	Data string
}

// Worker：工作的协程
func worker(id int, jobChan <-chan Job, resultChan chan<- string) {
	for job := range jobChan {
		fmt.Printf("Worker %d: 开始处理任务 %d\n", id, job.ID)
		time.Sleep(200 * time.Millisecond) // 模拟耗时
		resultChan <- fmt.Sprintf("Worker %d: 任务 %d 处理完毕", id, job.ID)
	}
}

func main() {
	// 1. 创建任务通道和结果通道
	jobChan := make(chan Job, 100) // 缓冲大一点，可以暂存任务
	resultChan := make(chan string, 100)

	// 2. 启动 Worker Pool（固定 3 个 Worker）
	for w := 1; w <= 3; w++ {
		go worker(w, jobChan, resultChan)
	}

	// 3. 发送 10 个任务（模拟高并发请求）
	go func() {
		for i := 1; i <= 10; i++ {
			job := Job{ID: i, Data: fmt.Sprintf("Traffic-Log-%d", i)}
			jobChan <- job
		}
		close(jobChan) // 发送完毕，关闭通道
	}()

	// 4. 收集结果
	for i := 1; i <= 10; i++ {
		fmt.Println(<-resultChan)
	}
}

package main

import (
	"fmt"
	"time"
)

// 模拟远程调用
func queryRemoteNode(nodeName string, success bool) <-chan string {
	ch := make(chan string)
	go func() {
		if success {
			time.Sleep(200 * time.Millisecond) // 正常响应
			ch <- fmt.Sprintf("%s 返回数据: 500MB", nodeName)
		} else {
			time.Sleep(2 * time.Second) // 模拟卡顿
			ch <- fmt.Sprintf("%s 终于响应了", nodeName)
		}
	}()
	return ch
}

func main() {
	fmt.Println("--- 测试正常节点 ---")
	doQuery("Node-A", true)
	
	fmt.Println("\n--- 测试超时节点 (500ms 超时) ---")
	doQuery("Node-B", false)
}

func doQuery(node string, success bool) {
	// 获取结果通道
	resultCh := queryRemoteNode(node, success)
	
	select {
	case res := <-resultCh:
		// 1. 正常收到数据
		fmt.Println("成功:", res)
	case <-time.After(500 * time.Millisecond):
		// 2. 超时触发
		fmt.Println("超时:", node, " 响应太慢，已断开！")
	}
}