浅谈Go语言中高效并发模式

go func() {
    fmt.Println("Hello from goroutine")
}()

ch := make(chan int)
go func() { ch <- 42 }()
value := <-ch
fmt.Println(value)

select {
case msg := <-ch1:
    fmt.Println("Got from ch1:", msg)
case msg := <-ch2:
    fmt.Println("Got from ch2:", msg)
case <-time.After(1 * time.Second):
    fmt.Println("Timeout")
}

package main
import (
	"fmt"
	"sync"
	"time"
)
func worker(id int, jobs <-chan int, results chan<- int, wg *sync.WaitGroup) {
	defer wg.Done()
	for j := range jobs {
		fmt.Printf("Worker %d started job %d\n", id, j)
		time.Sleep(time.Second) // 模拟耗时任务
		fmt.Printf("Worker %d finished job %d\n", id, j)
		results <- j * 2
	}
}
func main() {
	const numJobs = 5
	const numWorkers = 3
	jobs := make(chan int, numJobs)
	results := make(chan int, numJobs)
	var wg sync.WaitGroup
	// 启动3个worker
	for w := 1; w <= numWorkers; w++ {
		wg.Add(1)
		go worker(w, jobs, results, &wg)
	}
	// 发送5个任务
	for j := 1; j <= numJobs; j++ {
		jobs <- j
	}
	close(jobs)
	// 等待所有worker完成
	wg.Wait()
	close(results)
	// 收集结果
	for result := range results {
		fmt.Println("Result:", result)
	}
}

package main
import (
	"fmt"
	"sync"
	"time"
)
func producer(id int, out chan<- int, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 0; i < 3; i++ {
		fmt.Printf("Producer %d produced %d\n", id, i)
		out <- id*10 + i
		time.Sleep(200 * time.Millisecond)
	}
}
func consumer(in <-chan int, done chan<- bool) {
	for value := range in {
		fmt.Printf("Consumed: %d\n", value)
	}
	done <- true
}
func main() {
	ch := make(chan int)
	done := make(chan bool)
	var wg sync.WaitGroup
	// 启动3个producer
	for i := 1; i <= 3; i++ {
		wg.Add(1)
		go producer(i, ch, &wg)
	}
	// 启动1个consumer
	go consumer(ch, done)
	// 等待所有producer完成
	go func() {
		wg.Wait()
		close(ch)
	}()
	// 等待consumer完成
	<-done
	fmt.Println("All done!")
}

package main
import (
	"fmt"
	"sync"
)
func generator(nums ...int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for _, n := range nums {
			out <- n
		}
	}()
	return out
}
func square(in <-chan int) <-chan int {
	out := make(chan int)
	go func() {
		defer close(out)
		for n := range in {
			out <- n * n
		}
	}()
	return out
}
func main() {
	// 构建pipeline: generator -> square -> print
	input := generator(1, 2, 3, 4, 5)
	squares := square(input)
	// 打印结果
	for result := range squares {
		fmt.Println(result)
	}
}

package main
import (
	"context"
	"fmt"
	"time"
)
func longRunningTask(ctx context.Context) {
	select {
	case <-time.After(3 * time.Second):
		fmt.Println("Task completed")
	case <-ctx.Done():
		fmt.Println("Task canceled:", ctx.Err())
	}
}
func main() {
	// 带超时的context
	ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
	defer cancel()
	go longRunningTask(ctx)
	// 等待足够长时间观察结果
	time.Sleep(4 * time.Second)
}

package main
import (
	"fmt"
	"time"
)
type Future struct {
	result chan int
}
func NewFuture() (*Future, func(int)) {
	f := &Future{result: make(chan int)}
	return f, func(r int) { f.result <- r }
}
func (f *Future) Get() int {
	return <-f.result
}
func asyncCompute(complete func(int)) {
	go func() {
		time.Sleep(2 * time.Second)
		complete(42)
	}()
}
func main() {
	future, complete := NewFuture()
	asyncCompute(complete)
	fmt.Println("Waiting for result...")
	result := future.Get()
	fmt.Println("Result:", result)
}

package main
import (
	"fmt"
	"time"
)
func orDone(channels ...<-chan interface{}) <-chan interface{} {
	switch len(channels) {
	case 0:
		return nil
	case 1:
		return channels[0]
	}
	orDone := make(chan interface{})
	go func() {
		defer close(orDone)
		switch len(channels) {
		case 2:
			select {
			case <-channels[0]:
			case <-channels[1]:
			}
		default:
			select {
			case <-channels[0]:
			case <-channels[1]:
			case <-channels[2]:
			case <-orDone(channels[3:]...):
			}
		}
	}()
	return orDone
}
func main() {
	sig := make(chan interface{})
	time.AfterFunc(2*time.Second, func() { close(sig) })
	start := time.Now()
	<-orDone(sig, time.After(5*time.Second))
	fmt.Printf("Done after %v\n", time.Since(start))
}

package main
import (
	"fmt"
	"time"
)
func main() {
	requests := make(chan int, 5)
	for i := 1; i <= 5; i++ {
		requests <- i
	}
	close(requests)
	// 限制每200ms处理一个请求
	limiter := time.Tick(200 * time.Millisecond)
	for req := range requests {
		<-limiter
		fmt.Println("Request", req, time.Now())
	}
	// 突发限制模式
	burstyLimiter := make(chan time.Time, 3)
	for i := 0; i < 3; i++ {
		burstyLimiter <- time.Now()
	}
	go func() {
		for t := range time.Tick(200 * time.Millisecond) {
			burstyLimiter <- t
		}
	}()
	burstyRequests := make(chan int, 5)
	for i := 1; i <= 5; i++ {
		burstyRequests <- i
	}
	close(burstyRequests)
	for req := range burstyRequests {
		<-burstyLimiter
		fmt.Println("Bursty request", req, time.Now())
	}
}