package main

type TrafficData struct {
	SourceID string
	Count    int64
}

package main

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

// 模擬從數(shù)據(jù)源讀取數(shù)據(jù)（耗時操作）
func fetchTrafficFromSource(sourceID string) int64 {
	// 模擬網(wǎng)絡(luò)延遲 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) // 計數(shù)器 +1
		go func(id string) {
			defer wg.Done() // 協(xié)程結(jié)束時計數(shù)器 -1

			// 1. 獲取數(shù)據(jù)
			data := fetchTrafficFromSource(id)

			// 2. 寫入全局變量（加鎖保護）
			mu.Lock()
			totalTraffic += data
			mu.Unlock()
			
			fmt.Printf("Source %s 處理完成\n", id)
		}(source)
	}

	wg.Wait() // 等待所有協(xié)程結(jié)束

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

package main

import (
	"fmt"
	"time"
)

// 生產(chǎn)者：只負責讀數(shù)據(jù)，扔進通道
func producer(id string, ch chan<- TrafficData) {
	data := TrafficData{
		SourceID: id,
		Count:    fetchTrafficFromSource(id),
	}
	ch <- data // 發(fā)送數(shù)據(jù)到通道
	fmt.Printf("Producer: %s 發(fā)送數(shù)據(jù)\n", id)
}

// 消費者：只負責從通道拿數(shù)據(jù)并匯總
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"}
	
	// 創(chuàng)建一個帶緩沖的通道，緩沖大小為 5
	// 緩沖通道可以協(xié)程解耦，生產(chǎn)者不需要阻塞等待消費者接收
	dataCh := make(chan TrafficData, 5) 
	doneCh := make(chan bool)

	startTime := time.Now()

	// 啟動消費者
	go consumer(dataCh, doneCh)

	// 啟動生產(chǎn)者
	for _, source := range sources {
		go producer(source, dataCh)
	}

	// 監(jiān)控：當所有生產(chǎn)者都結(jié)束后，關(guān)閉通道
	// 注意：實際項目中通常用 WaitGroup 來協(xié)調(diào)，這里為了簡化邏輯
	time.Sleep(1 * time.Second) 
	close(dataCh) // 關(guān)閉通道，消費者會結(jié)束 for range 循環(huán)

	<-doneCh // 等待消費者結(jié)束
	fmt.Printf("總耗時: %v\n", time.Since(startTime))
}

package main

import (
	"fmt"
	"time"
)

// 任務(wù)：包含任務(wù) ID 和處理邏輯
type Job struct {
	ID   int
	Data string
}

// Worker：工作的協(xié)程
func worker(id int, jobChan <-chan Job, resultChan chan<- string) {
	for job := range jobChan {
		fmt.Printf("Worker %d: 開始處理任務(wù) %d\n", id, job.ID)
		time.Sleep(200 * time.Millisecond) // 模擬耗時
		resultChan <- fmt.Sprintf("Worker %d: 任務(wù) %d 處理完畢", id, job.ID)
	}
}

func main() {
	// 1. 創(chuàng)建任務(wù)通道和結(jié)果通道
	jobChan := make(chan Job, 100) // 緩沖大一點，可以暫存任務(wù)
	resultChan := make(chan string, 100)

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

	// 3. 發(fā)送 10 個任務(wù)（模擬高并發(fā)請求）
	go func() {
		for i := 1; i <= 10; i++ {
			job := Job{ID: i, Data: fmt.Sprintf("Traffic-Log-%d", i)}
			jobChan <- job
		}
		close(jobChan) // 發(fā)送完畢，關(guān)閉通道
	}()

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

package main

import (
	"fmt"
	"time"
)

// 模擬遠程調(diào)用
func queryRemoteNode(nodeName string, success bool) <-chan string {
	ch := make(chan string)
	go func() {
		if success {
			time.Sleep(200 * time.Millisecond) // 正常響應(yīng)
			ch <- fmt.Sprintf("%s 返回數(shù)據(jù): 500MB", nodeName)
		} else {
			time.Sleep(2 * time.Second) // 模擬卡頓
			ch <- fmt.Sprintf("%s 終于響應(yīng)了", nodeName)
		}
	}()
	return ch
}

func main() {
	fmt.Println("--- 測試正常節(jié)點 ---")
	doQuery("Node-A", true)
	
	fmt.Println("\n--- 測試超時節(jié)點 (500ms 超時) ---")
	doQuery("Node-B", false)
}

func doQuery(node string, success bool) {
	// 獲取結(jié)果通道
	resultCh := queryRemoteNode(node, success)
	
	select {
	case res := <-resultCh:
		// 1. 正常收到數(shù)據(jù)
		fmt.Println("成功:", res)
	case <-time.After(500 * time.Millisecond):
		// 2. 超時觸發(fā)
		fmt.Println("超時:", node, " 響應(yīng)太慢，已斷開！")
	}
}