基本语法

创建channel

ch1 := make(chan struct{})
ch2 := make(chan int, 10)

读写channel

从channel中读取数据叫做recv；向channel写数据叫做send。

// 发送数据到channel
ch1 <- struct{}{}
ch2 <- 10
// 从channel接收数据
val1 := <-ch1
// ok可以判断channel是否已经关闭
val2, ok := <-ch2

遍历channel

该语法仅限于带缓冲区的channel

package main

import "fmt"

func main() {
	ch := make(chan int, 2)
	go func() {
		for i := 1; i < 10; i++ {
			ch <- i
		}
		close(ch)
	}()
	for val := range ch {
		fmt.Printf("%v ", val)
	}
	fmt.Println()
}

channel的状态与读写关系

惯用技法

协程池

package main

import (
	"fmt"
	"time"
)

type Worker struct {
	Name string
}

func (w *Worker) ProcessTask(data string, done chan<- *Worker) {
	time.Sleep(time.Second)
	fmt.Println(w.Name + " task done:" + data)
	done <- w
}

type WorkerPool struct {
	taskCh  chan string
	done    chan *Worker
	closeCh chan struct{}
	pool    []*Worker
}

func NewWorkerPool(workCnt int) *WorkerPool {
	wp := &WorkerPool{
		taskCh:  make(chan string),
		done:    make(chan *Worker),
		closeCh: make(chan struct{}),
		pool:    make([]*Worker, workCnt),
	}
	for i := 0; i < len(wp.pool); i++ {
		wp.pool[i] = &Worker{Name: fmt.Sprintf("worker-%v", i)}
	}
	wp.Run()
	return wp
}

func (wp *WorkerPool) AddTask(data string) {
	wp.taskCh <- data
}

func (wp *WorkerPool) Run() {
	go func() {
		for {
			select {
			case w := <-wp.done:
				wp.pool = append(wp.pool, w)
			case <-wp.closeCh:
				close(wp.done)
				close(wp.taskCh)
				return
			default:
				if len(wp.pool) > 0 {
					w := wp.pool[0]
					wp.pool = wp.pool[1:]
					go w.ProcessTask(<-wp.taskCh, wp.done)
				}
			}
		}
	}()
}

func (wp *WorkerPool) ShutDown() {
	time.Sleep(time.Second * 3)
	close(wp.closeCh)
}

func generator() chan string {
	inputCh := make(chan string)
	go func() {
		taskID := 1
		for {
			inputCh <- fmt.Sprintf("task-%v", taskID)
			taskID++
			if taskID == 21 {
				close(inputCh)
				break
			}
		}
	}()
	return inputCh
}

func main() {
	wp := NewWorkerPool(5)
	for task := range generator() {
		wp.AddTask(task)
	}
	wp.ShutDown()
	fmt.Println("all task done")
}