import "golang.org/x/sync/semaphore"

// 创建一个容量为 10 的信号量
sem := semaphore.NewWeighted(10)

// 获取 1 个资源（阻塞直到有可用资源）
sem.Acquire(ctx, 1)

// 释放 1 个资源
sem.Release(1)

// 尝试获取（非阻塞，返回 bool）
if sem.TryAcquire(3) {
    // 成功获取 3 个资源
}

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"golang.org/x/sync/semaphore"
)

func main() {
	const maxConcurrency = 3
	sem := semaphore.NewWeighted(maxConcurrency)
	var wg sync.WaitGroup

	for i := 0; i < 10; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			// 获取信号量，超时 5 秒
			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
			defer cancel()

			if err := sem.Acquire(ctx, 1); err != nil {
				fmt.Printf("Worker %d: 获取信号量超时\n", id)
				return
			}
			defer sem.Release(1)

			fmt.Printf("Worker %d: 开始（并发数 ≤ %d）\n", id, maxConcurrency)
			time.Sleep(time.Second) // 模拟工作
		}(i)
	}
	wg.Wait()
}

import "golang.org/x/sync/singleflight"

var g singleflight.Group

// Do：同 key 并发只执行一次
v, err, shared := g.Do("cache-key", func() (any, error) {
    // 只有一个 goroutine 会执行这个函数
    return queryDB("cache-key")
})
// shared == true 表示结果是和其他 goroutine 共享的

// DoChan：异步版本，返回 channel
ch := g.DoChan("cache-key", func() (any, error) {
    return queryDB("cache-key")
})
result := <-ch

// Forget：删除某个 key，下一次调用会重新执行
g.Forget("cache-key")

var g singleflight.Group
var cache sync.Map

func getData(key string) (string, error) {
	// 1. 先查缓存
	if v, ok := cache.Load(key); ok {
		return v.(string), nil
	}

	// 2. 缓存未命中，用 SingleFlight 合并请求
	v, err, _ := g.Do(key, func() (any, error) {
		// 只有一个 goroutine 会真正查 DB
		result, err := queryDB(key)
		if err != nil {
			return nil, err
		}
		cache.Store(key, result) // 回填缓存
		return result, nil
	})
	if err != nil {
		return "", err
	}
	return v.(string), nil
}

import "golang.org/x/sync/errgroup"

g, ctx := errgroup.WithContext(context.Background())

g.Go(func() error {
    return fetchUsers(ctx)
})
g.Go(func() error {
    return fetchOrders(ctx)
})
g.Go(func() error {
    return fetchProducts(ctx)
})

if err := g.Wait(); err != nil {
    log.Fatal("子任务失败:", err)
}

g, ctx := errgroup.WithContext(context.Background())
g.SetLimit(5) // 最多 5 个子任务同时执行

for _, url := range urls {
    u := url
    g.Go(func() error {
        return fetch(ctx, u) // 超过 5 个并发时会阻塞
    })
}

if err := g.Wait(); err != nil {
    log.Fatal(err)
}

import "github.com/marusama/cyclicbarrier"

// 创建一个 3 参与者的栅栏
b := cyclicbarrier.New(3)

for i := 0; i < 3; i++ {
    go func(id int) {
        for round := 0; round < 5; round++ {
            fmt.Printf("Worker %d: 第 %d 轮准备就绪\n", id, round)
            b.Await(context.Background()) // 等待所有人到达
            // 全部到达后继续，栅栏自动重置
            fmt.Printf("Worker %d: 第 %d 轮开始执行\n", id, round)
        }
    }(i)
}

import (
    clientv3 "go.etcd.io/etcd/client/v3"
    "go.etcd.io/etcd/client/v3/concurrency"
)

// 连接 etcd
cli, _ := clientv3.New(clientv3.Config{
    Endpoints: []string{"localhost:2379"},
})
defer cli.Close()

// 创建 session（带 TTL 的租约，节点宕机后自动释放）
session, _ := concurrency.NewSession(cli, concurrency.WithTTL(10))
defer session.Close()

// 参与选举
e := concurrency.NewElection(session, "/my-service/leader")

// 竞选 Leader（阻塞直到当选）
if err := e.Campaign(ctx, "node-1"); err != nil {
    log.Fatal(err)
}
fmt.Println("我是 Leader！")

// 查看当前 Leader
resp, _ := e.Leader(ctx)
fmt.Println("当前 Leader:", string(resp.Kvs[0].Value))

// 主动让位
e.Resign(ctx)

session, _ := concurrency.NewSession(cli)
mu := concurrency.NewMutex(session, "/my-lock/")

// 获取锁（阻塞直到成功）
if err := mu.Lock(ctx); err != nil {
    log.Fatal(err)
}
fmt.Println("获取到分布式锁")

// 临界区操作
doSomething()

// 释放锁
if err := mu.Unlock(ctx); err != nil {
    log.Fatal(err)
}

session, _ := concurrency.NewSession(cli)
rw := recipe.NewRWMutex(session, "/my-rwlock/")

// 读锁（多个节点可以同时持有）
rw.RLock(ctx)
readData()
rw.RUnlock(ctx)

// 写锁（独占）
rw.Lock(ctx)
writeData()
rw.Unlock(ctx)

import recipe "go.etcd.io/etcd/contrib/recipes"

q := recipe.NewQueue(cli, "/my-queue")

// 入队（可以在任意节点）
q.Enqueue("task-1")
q.Enqueue("task-2")

// 出队（可以在另一个节点）
val, _ := q.Dequeue() // "task-1"（FIFO 顺序）
// 如果队列为空，Dequeue 会阻塞等待

pq := recipe.NewPriorityQueue(cli, "/my-pqueue")

// 入队，附带优先级（数字越小优先级越高）
pq.Enqueue("low-priority-task", 10)
pq.Enqueue("high-priority-task", 1)

val, _ := pq.Dequeue() // "high-priority-task"（优先级 1 先出）

b := recipe.NewBarrier(cli, "/my-barrier")

// 节点 A：创建栅栏（阻塞其他节点）
b.Hold()

// 节点 B/C/D：在栅栏处等待
b.Wait() // 阻塞，直到栅栏被移除

// 节点 A：移除栅栏（放行所有等待者）
b.Release()

db := recipe.NewDoubleBarrier(session, "/my-dbarrier", 3) // 需要 3 个参与者

// 每个节点调用 Enter（阻塞直到 3 个节点都进入）
db.Enter() // 3 个节点都 Enter 后，全部放行

// 执行任务...

// 每个节点调用 Leave（阻塞直到 3 个节点都离开）
db.Leave() // 3 个节点都 Leave 后，全部继续

import "go.etcd.io/etcd/client/v3/concurrency"

// 转账：从 A 扣款 100，给 B 加款 100（原子操作）
_, err := concurrency.NewSTM(cli, func(stm concurrency.STM) error {
    balanceA := stm.Get("/account/A")
    balanceB := stm.Get("/account/B")

    a, _ := strconv.Atoi(balanceA)
    b, _ := strconv.Atoi(balanceB)

    if a < 100 {
        return fmt.Errorf("余额不足")
    }

    stm.Put("/account/A", strconv.Itoa(a-100))
    stm.Put("/account/B", strconv.Itoa(b+100))
    return nil
})

# 两个应该加入 CI 的命令
go vet ./...
go test -race ./...