NOTE

4.5 生产者消费者模式

1. 生产者/消费者模式是什么 - 有三个角色：生产者、消费者、队列 - 生产者生产完数据丢入队列，消费者从队列取出数据消费 2. 为什么需要生产者/消费者模式 - 解耦 - 生产者只关注生产数据，不关心消费者怎么消费 - 消费者只关注消费数据，不关心生产者怎么生产 - 异步 3. 生产者/消费者模

Software Architecture & Engineering创建于 2021/05/28更新于 2021/05/29historical

这是历史学习笔记，可能存在过时或不完整的理解。

1. 生产者/消费者模式是什么

有三个角色：生产者、消费者、队列
- 生产者生产完数据丢入队列，消费者从队列取出数据消费

2. 为什么需要生产者/消费者模式

解耦
- 生产者只关注生产数据，不关心消费者怎么消费
- 消费者只关注消费数据，不关心生产者怎么生产
异步

3. 生产者/消费者模式实现

3.1. Golang

模仿Java BlockingQueue

如果n为1那么适合生产者消费者速度匹配的场景
如果n>1那么适合生产者快，消费者慢的场景

type Config struct {
	ID int
}

func (c *Config) String() string {
	return fmt.Sprintf("Config[ID=%v]", c.ID)

}

type BlockingQueue struct {
	pipe chan *Config
}

func NewBlockingQueue(n int) *BlockingQueue {
	return &BlockingQueue{make(chan *Config, n)}
}

func (b *BlockingQueue) Offer(data *Config) {
	b.pipe <- data
}

func (b *BlockingQueue) Take() (*Config, bool) {
	saasConfig, ok := <-b.pipe
	return saasConfig, ok
}

func (b *BlockingQueue) Close() {
	close(b.pipe)
}

func TestMain3(t *testing.T) {
	queue := NewBlockingQueue(10)

	go produce(queue)
	go consume(queue)

	time.Sleep(time.Hour)
}

func consume(queue *BlockingQueue) {
	for {
		saasConfig, ok := queue.Take()
		if !ok {
			break
		}
		fmt.Println(saasConfig)
	}
}

func produce(queue *BlockingQueue) {
	defer queue.Close()
	for i := 0; i < 100; i++ {
		time.Sleep(time.Microsecond * 200)
		queue.Offer(&Config{ID: i})
	}
}

增加超时

type Config struct {
	ID int
}

func (c *Config) String() string {
	return fmt.Sprintf("Config[ID=%v]", c.ID)

}

func TestMain3(t *testing.T) {
	//设置一秒后超时的context
	d := time.Now().Add(1 * time.Second)
	ctx, cancel := context.WithDeadline(context.Background(), d)
	defer cancel()

	channel := make(chan *Config, 10)
	go produce(ctx, channel)
	go consume(ctx, channel)

	time.Sleep(time.Hour)
}

func consume(ctx context.Context, queue chan *Config) {
	defer fmt.Println("消费者退出")
	for {
		select {
		case cfg, ok := <-queue:
			if !ok {
				fmt.Println("消费者收到关闭信号")
				return
			}
			handleCfg(cfg)
		case <-ctx.Done():
			fmt.Println("消费者超时：err=", ctx.Err())
			return
		}
	}
}

func handleCfg(cfg *Config) {
	fmt.Println(cfg)
}

func produce(ctx context.Context, queue chan *Config) {
	defer fmt.Println("生产者退出")
	defer close(queue)
	defer fmt.Println("生产者发送关闭信号")
	for i := 0; i < 100; i++ {
		time.Sleep(time.Millisecond * 200)
		c := fetchCfg(i)
		select {
		case queue <- c:
		case <-ctx.Done():
			fmt.Println("生产者超时：err=", ctx.Err())
			return
		}
	}
}

func fetchCfg(i int) *Config {
	return &Config{ID: i}
}

面向对象


type Config struct {
	ID int
}

func (c *Config) String() string {
	return fmt.Sprintf("Config[ID=%v]", c.ID)
}

type ConfigHandler struct {
	queue chan *Config
}

func NewConfigHandler(num int) *ConfigHandler {
	return &ConfigHandler{queue: make(chan *Config, num)}
}

func (c *ConfigHandler) produce(ctx context.Context) {
	defer fmt.Println("生产者退出")
	defer close(c.queue)
	defer fmt.Println("生产者发送关闭信号")
	for i := 0; i < 100; i++ {
		time.Sleep(time.Millisecond * 200)
		cfg := fetchCfg(i)
		select {
		case c.queue <- cfg:
		case <-ctx.Done():
			fmt.Println("生产者超时：err=", ctx.Err())
			return
		}
	}
}

func (c *ConfigHandler) consume(ctx context.Context) {
	defer fmt.Println("消费者退出")
	for {
		select {
		case cfg, ok := <-c.queue:
			if !ok {
				fmt.Println("消费者收到关闭信号")
				return
			}
			handleCfg(cfg)
		case <-ctx.Done():
			fmt.Println("消费者超时：err=", ctx.Err())
			return
		}
	}
}

func TestMain3(t *testing.T) {
	//设置一秒后超时的context
	d := time.Now().Add(1 * time.Second)
	ctx, cancel := context.WithDeadline(context.Background(), d)
	defer cancel()

	handler := NewConfigHandler(10)

	go handler.produce(ctx)
	go handler.consume(ctx)

	time.Sleep(time.Hour)
}

func handleCfg(cfg *Config) {
	fmt.Println(cfg)
}

func fetchCfg(i int) *Config {
	return &Config{ID: i}
}

增加错误处理

type Config struct {
	ID int
}

func (c *Config) String() string {
	return fmt.Sprintf("Config[ID=%v]", c.ID)
}

type ConfigHandler struct {
	queue chan *Config
}

func NewConfigHandler(num int) *ConfigHandler {
	return &ConfigHandler{queue: make(chan *Config, num)}
}

func (c *ConfigHandler) produce(ctx context.Context) error {
	defer fmt.Println("生产者退出")
	defer close(c.queue)
	defer fmt.Println("生产者发送关闭信号")
	for i := 0; i < 100; i++ {
		time.Sleep(time.Millisecond * 200)
		cfg, err := fetchCfg(i)
		if err != nil {
			continue
		}
		select {
		case c.queue <- cfg:
		case <-ctx.Done():
			fmt.Println("生产者超时：err=", ctx.Err())
			return ctx.Err()
		}
	}
	return nil
}

func (c *ConfigHandler) consume(ctx context.Context) error {
	defer fmt.Println("消费者退出")
	for {
		select {
		case cfg, ok := <-c.queue:
			if !ok {
				fmt.Println("消费者收到关闭信号")
				return nil
			}
			err := handleCfg(cfg)
			if err != nil {
				return err
			}
		case <-ctx.Done():
			fmt.Println("消费者超时：err=", ctx.Err())
			return ctx.Err()
		}
	}
}

func TestMain3(t *testing.T) {
	//设置一秒后超时的context
	ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(10*time.Second))
	defer cancel()

	handler := NewConfigHandler(10)

	group, newCtx := errgroup.WithContext(ctx)
	group.Go(func() error {
		return handler.produce(newCtx)
	})
	group.Go(func() error {
		return handler.consume(newCtx)
	})
	group.Go(func() error {
	return handler.consume(newCtx)
})
	err := group.Wait()
	if err != nil {
		fmt.Println("WaitGroup: err=", err)
	}
	fmt.Println("main done")
}

func handleCfg(cfg *Config) error {
	fmt.Println(cfg)
	if cfg.ID == 20 {
		return fmt.Errorf("消费者处理出错")
	}
	return nil
}

func fetchCfg(i int) (*Config, error) {
	return &Config{ID: i}, nil
}

4. 参考