foo

a.go

package main

import (
	"bytes"
	"fmt"
	"io"
	"io/ioutil"
	"sync"
)

func main() {
	// 输入，可以是任意 io.Reader
	input := bytes.NewReader([]byte("foobarbaz"))

	// 最大 4 个处理线程
	sem := make(chan struct{}, 4)
	cond := sync.NewCond(new(sync.Mutex))
	next := 0
	ready := 0
	for {
		// 分段读取
		r := &io.LimitedReader{
			R: input,
			N: 2,
		}
		chunk, err := ioutil.ReadAll(r)

		if len(chunk) > 0 {
			sem <- struct{}{}
			i := next
			next++
			go func() {
				defer func() {
					<-sem
				}()

				// 按顺序输出
				cond.L.Lock()
				for i != ready {
					cond.Wait()
				}
				fmt.Printf("%s\n", chunk)
				ready++
				cond.Broadcast()
				cond.L.Unlock()

			}()
		}

		if err == io.EOF {
			break
		} else if err != nil {
			panic(err)
		}
		if len(chunk) == 0 {
			break
		}
	}
	for i := 0; i < cap(sem); i++ {
		sem <- struct{}{}
	}

}