更友好的并发库conc介绍

// waitGroupDemo conc.WaitGroup并发示例
func waitGroupDemo() {
	var count atomic.Int64

	var wg conc.WaitGroup
	// 开启10个goroutine并发执行 count.Add(1)
	for i := 0; i < 10; i++ {
		wg.Go(func() {
			count.Add(1)
		})
	}
	// 等待10个goroutine都执行完
	wg.Wait()

	fmt.Println(count.Load())
}

// waitGroupDemo 自动recover示例
func waitGroupDemo2() {
	var count atomic.Int64

	var wg conc.WaitGroup
	// 开启10个goroutine并发执行 count.Add(1)
	for i := 0; i < 10; i++ {
		wg.Go(func() {
			if i == 7 {
				panic("bad thing")
			}
			count.Add(1)
		})
	}
	// 等待10个goroutine都执行完
	wg.WaitAndRecover()

	fmt.Println(count.Load())
}

// poolDemo goroutine池示例
func poolDemo() {
	// 创建一个最大数量为3的goroutine池
	p := pool.New().WithMaxGoroutines(3)
	// 使用p.Go()提交5个任务
	for i := 0; i < 5; i++ {
		p.Go(func() {
			fmt.Println("Q1mi")
		})
	}
	p.Wait()
}

// poolWithContextDemoCancelOnError 支持context的池
// goroutine中出错时取消context
func poolWithContextDemoCancelOnError() {
	p := pool.New().
		WithMaxGoroutines(4).
		WithContext(context.Background()).
		WithCancelOnError() // 出错时取消所有goroutine
	// 提交3个任务
	for i := 0; i < 3; i++ {
		i := i
		p.Go(func(ctx context.Context) error {
			if i == 2 {
				return errors.New("cancel all other tasks")
			}
			<-ctx.Done()
			return nil
		})
	}
	err := p.Wait()
	fmt.Println(err)
}

func poolWithError() {
	p := pool.New().WithErrors()
	for i := 0; i < 3; i++ {
		i := i
		p.Go(func() error {
			if i == 2 {
				return errors.New("oh no!")
			}
			return nil
		})
	}
	err := p.Wait()
	fmt.Println(err)
}

// poolWithResult 执行返回结果的任务池
func poolWithResult() {
	// 创建一个任务池，其中任务返回的结果为int
	p := pool.NewWithResults[int]()
	for i := 0; i < 10; i++ {
		i := i
		p.Go(func() int {
			return i * 2
		})
	}
	res := p.Wait()
	// 结果的顺序是不确定的, 所以这里先排序再打印
	sort.Ints(res)
	fmt.Println(res)
}

// streamDemo 并发的流式任务示例
func streamDemo() {
	times := []int{20, 52, 16, 45, 4, 80}

	s := stream.New()
	for _, millis := range times {
		dur := time.Duration(millis) * time.Millisecond
		// 提交任务
		s.Go(func() stream.Callback {
			time.Sleep(dur)
			// 虽然上一行通过sleep增加了时间
			// 但最终结果仍按任务提交（s.Go）的顺序打印
			return func() { fmt.Println(dur) }
		})
	}
	s.Wait()
}

20ms
52ms
16ms
45ms
4ms
80ms

func ForEach[T any](input []T, f func(*T))
func ForEachIdx[T any](input []T, f func(int, *T))
func Map[T, R any](input []T, f func(*T) R) []R
func MapErr[T, R any](input []T, f func(*T) (R, error)) ([]R, error)

// iterDemo 迭代器ForEach示例
func iterDemo() {
	input := []int{1, 2, 3, 4}
	// 可直接调用iter包的ForEach函数
	iter.ForEach(input, func(v *int) {
		if *v%2 != 0 {
			*v = -1
		}
	})
	fmt.Println(input)
}

// iteratorDemo 迭代器示例
func iteratorDemo() {
	input := []int{1, 2, 3, 4}
	// 创建一个最大goroutine个数为输入元素一半的迭代器
	iterator := iter.Iterator[int]{
		MaxGoroutines: len(input) / 2,
	}

	iterator.ForEach(input, func(v *int) {
		if *v%2 != 0 {
			*v = -1
		}
	})

	fmt.Println(input)
}

[-1 2 -1 4]

// mapperDemo mapper示例
func mapperDemo() {
	input := []int{1, 2, 3, 4}
	// 创建一个最大goroutine个数为输入元素一半的映射器
	mapper := iter.Mapper[int, bool]{
		MaxGoroutines: len(input) / 2,
	}

	results := mapper.Map(input, func(v *int) bool { return *v%2 == 0 })
	fmt.Println(results)
}

[false true false true]

// panicDemo recover可能出现的异常
func panicDemo() {
	var pc panics.Catcher
	i := 0
	pc.Try(func() { i += 1 })
	pc.Try(func() { panic("abort!") })
	pc.Try(func() { i += 1 })

	// recover可能出现的panic
	rc := pc.Recovered()
	// 重新panic
	// pc.Repanic()

	fmt.Println(i)
	fmt.Println(rc.Value.(string))
}