Go Select 与并发模式

package main

import (
	"fmt"
	"time"
)

// 模拟一个可能超时的操作
func slowOperation(ch chan string) {
	time.Sleep(2 * time.Second)
	ch <- "操作完成"
}

func main() {
	ch := make(chan string)

	go slowOperation(ch)

	// 设置 1 秒超时
	select {
	case result := <-ch:
		fmt.Println("成功:", result)
	case <-time.After(1 * time.Second):
		fmt.Println("操作超时！")
	}

	// 带超时的函数
	fmt.Println("\n--- 带超时的函数 ---")
	result, err := fetchWithTimeout("http://example.com", 500*time.Millisecond)
	if err != nil {
		fmt.Println("错误:", err)
	} else {
		fmt.Println("结果:", result)
	}
}

func fetchWithTimeout(url string, timeout time.Duration) (string, error) {
	ch := make(chan string, 1) // 有缓冲，防止 goroutine 泄漏

	go func() {
		// 模拟网络请求
		time.Sleep(1 * time.Second)
		ch <- "响应数据"
	}()

	select {
	case result := <-ch:
		return result, nil
	case <-time.After(timeout):
		return "", fmt.Errorf("请求 %s 超时（%v）", url, timeout)
	}
}

package main

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

// 生产者：每个 goroutine 产生一组数据
func producer(id int, count int) <-chan int {
	ch := make(chan int)
	go func() {
		for i := 0; i < count; i++ {
			ch <- id*100 + i
			time.Sleep(50 * time.Millisecond)
		}
		close(ch)
	}()
	return ch
}

// merge 将多个 channel 合并为一个（扇入）
func merge(channels ...<-chan int) <-chan int {
	var wg sync.WaitGroup
	out := make(chan int)

	// 将每个 channel 的数据复制到 out
	output := func(ch <-chan int) {
		defer wg.Done()
		for val := range ch {
			out <- val
		}
	}

	for _, ch := range channels {
		wg.Add(1)
		go output(ch)
	}

	// 所有输入 channel 关闭后，关闭输出 channel
	go func() {
		wg.Wait()
		close(out)
	}()

	return out
}

func main() {
	// 创建 3 个生产者
	ch1 := producer(1, 3)
	ch2 := producer(2, 3)
	ch3 := producer(3, 3)

	// 扇入合并
	merged := merge(ch1, ch2, ch3)

	// 消费合并后的数据
	for val := range merged {
		fmt.Printf("收到: %d\n", val)
	}
	fmt.Println("所有生产者数据已合并处理完毕")
}

package main

import (
	"fmt"
	"math"
)

// 阶段 1：生成数字
func generate(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		for _, n := range nums {
			out <- n
		}
		close(out)
	}()
	return out
}

// 阶段 2：计算平方
func square(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			out <- n * n
		}
		close(out)
	}()
	return out
}

// 阶段 3：过滤大于 10 的值
func filter(in <-chan int, threshold int) <-chan int {
	out := make(chan int)
	go func() {
		for n := range in {
			if n > threshold {
				out <- n
			}
		}
		close(out)
	}()
	return out
}

// 阶段 4：计算平方根
func sqrt(in <-chan int) <-chan float64 {
	out := make(chan float64)
	go func() {
		for n := range in {
			out <- math.Sqrt(float64(n))
		}
		close(out)
	}()
	return out
}

func main() {
	// 构建管道：生成 -> 平方 -> 过滤 -> 平方根
	nums := generate(1, 2, 3, 4, 5, 6, 7)
	squared := square(nums)
	filtered := filter(squared, 10)
	result := sqrt(filtered)

	// 消费最终结果
	fmt.Println("管道处理结果:")
	for val := range result {
		fmt.Printf("  %.2f\n", val)
	}
}

package main

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

// SearchResult 搜索结果
type SearchResult struct {
	Source string
	Data   string
}

// 模拟搜索不同数据源
func searchGoogle(query string, timeout time.Duration) SearchResult {
	time.Sleep(150 * time.Millisecond)
	return SearchResult{Source: "Google", Data: fmt.Sprintf("Google 结果: %s", query)}
}

func searchBing(query string, timeout time.Duration) SearchResult {
	time.Sleep(200 * time.Millisecond)
	return SearchResult{Source: "Bing", Data: fmt.Sprintf("Bing 结果: %s", query)}
}

func searchLocal(query string, timeout time.Duration) SearchResult {
	time.Sleep(50 * time.Millisecond)
	return SearchResult{Source: "Local", Data: fmt.Sprintf("本地结果: %s", query)}
}

// 并发搜索所有数据源，返回最先完成的结果
func concurrentSearch(query string) SearchResult {
	ch := make(chan SearchResult, 3) // 有缓冲避免 goroutine 泄漏

	go func() { ch <- searchGoogle(query, time.Second) }()
	go func() { ch <- searchBing(query, time.Second) }()
	go func() { ch <- searchLocal(query, time.Second) }()

	// 返回最先到达的结果
	return <-ch
}

// 并发搜索并收集所有结果
func searchAll(query string) []SearchResult {
	ch := make(chan SearchResult, 3)
	var wg sync.WaitGroup

	sources := []func(string, time.Duration) SearchResult{
		searchGoogle, searchBing, searchLocal,
	}

	for _, search := range sources {
		wg.Add(1)
		go func(fn func(string, time.Duration) SearchResult) {
			defer wg.Done()
			ch <- fn(query, time.Second)
		}(search)
	}

	go func() {
		wg.Wait()
		close(ch)
	}()

	var results []SearchResult
	for r := range ch {
		results = append(results, r)
	}
	return results
}

func main() {
	query := "Go 并发编程"

	fmt.Println("=== 最快结果 ===")
	fastest := concurrentSearch(query)
	fmt.Printf("[%s] %s\n\n", fastest.Source, fastest.Data)

	fmt.Println("=== 所有结果 ===")
	all := searchAll(query)
	for _, r := range all {
		fmt.Printf("[%s] %s\n", r.Source, r.Data)
	}
}