Go语言实现的可读性更高的并发神库详解

func main(){
    var wg sync.WaitGroup
    for i:=0; i 

func main() {
	wg := conc.NewWaitGroup()
	for i := 0; i < 10; i++ {
		wg.Go(doSomething)
	}
	wg.Wait()
}
func doSomething() {
	fmt.Println("test")
}

type WaitGroup struct {
	wg sync.WaitGroup
	pc panics.Catcher
}

type Catcher struct {
	recovered atomic.Pointer[RecoveredPanic]
}

type RecoveredPanic struct {
	// The original value of the panic.
	Value any
	// The caller list as returned by runtime.Callers when the panic was
	// recovered. Can be used to produce a more detailed stack information with
	// runtime.CallersFrames.
	Callers []uintptr
	// The formatted stacktrace from the goroutine where the panic was recovered.
	// Easier to use than Callers.
	Stack []byte
}

func (p *Catcher) Try(f func()) {
	defer p.tryRecover()
	f()
}
func (p *Catcher) tryRecover() {
	if val := recover(); val != nil {
		rp := NewRecoveredPanic(1, val)
        // 只会记录第一个panic的goroutine信息
		p.recovered.CompareAndSwap(nil, &rp)
	}
}

func (p *Catcher) Repanic() {
	if val := p.Recovered(); val != nil {
		panic(val)
	}
}
func (p *Catcher) Recovered() *RecoveredPanic {
	return p.recovered.Load()
}

func (h *WaitGroup) Wait() {
	h.wg.Wait()
	// Propagate a panic if we caught one from a child goroutine.
	h.pc.Repanic()
}
func (h *WaitGroup) WaitAndRecover() *panics.RecoveredPanic {
	h.wg.Wait()
	// Return a recovered panic if we caught one from a child goroutine.
	return h.pc.Recovered()
}

import "github.com/sourcegraph/conc/pool"
func ExampleContextPool_WithCancelOnError() {
	p := pool.New().
		WithMaxGoroutines(4).
		WithContext(context.Background()).
		WithCancelOnError()
	for i := 0; i 

type Pool struct {
	handle   conc.WaitGroup
	limiter  limiter
	tasks    chan func()
	initOnce sync.Once
}

type limiter chan struct{}
func (l limiter) limit() int {
	return cap(l)
}
func (l limiter) release() {
	if l != nil {
		<-l
	}
}

func (p *Pool) Go(f func()) {
	p.init()
	if p.limiter == nil {
		// 没有限制
		select {
		case p.tasks 

func ExampleStream() {
	times := []int{20, 52, 16, 45, 4, 80}
	stream := stream2.New()
	for _, millis := range times {
		dur := time.Duration(millis) * time.Millisecond
		stream.Go(func() stream2.Callback {
			time.Sleep(dur)
			// This will print in the order the tasks were submitted
			return func() { fmt.Println(dur) }
		})
	}
	stream.Wait()
	// Output:
	// 20ms
	// 52ms
	// 16ms
	// 45ms
	// 4ms
	// 80ms
}

type Stream struct {
	pool             pool.Pool
	callbackerHandle conc.WaitGroup
	queue            chan callbackCh
	initOnce sync.Once
}

type callbackCh chan func()

func (s *Stream) Go(f Task) {
	s.init()
	// Get a channel from the cache.
	ch := getCh()
	// Queue the channel for the callbacker.
	s.queue 

func main() {
	input := []int{1, 2, 3, 4}
	iterator := iter.Iterator[int]{
		MaxGoroutines: len(input) / 2,
	}
	iterator.ForEach(input, func(v *int) {
		if *v%2 != 0 {
			*v = -1
		}
	})
	fmt.Println(input)
}

type Iterator[T any] struct {
	MaxGoroutines int
}
func (iter Iterator[T]) ForEachIdx(input []T, f func(int, *T)) {
	if iter.MaxGoroutines == 0 {
		// iter is a value receiver and is hence safe to mutate
		iter.MaxGoroutines = defaultMaxGoroutines()
	}
	numInput := len(input)
	if iter.MaxGoroutines > numInput {
		// No more concurrent tasks than the number of input items.
		iter.MaxGoroutines = numInput
	}
	var idx atomic.Int64
	// 通过atomic控制仅创建一个闭包
	task := func() {
		i := int(idx.Add(1) - 1)
		for ; i 

func main() {
	input := []int{1, 2, 3, 4}
	mapper := iter.Mapper[int, bool]{
		MaxGoroutines: len(input) / 2,
	}
	results := mapper.Map(input, func(v *int) bool { return *v%2 == 0 })
	fmt.Println(results)
	// Output:
	// [false true false true]
}