Go并发编程— —Timer、Ticker、WaitGroup及其他常用模型

我不是女神ヾ 2024-03-16 16:39 57阅读 0赞

## Go并发编程— —Timer、Ticker、WaitGroup用法 ##

### 1 Timer（执行一次） ###

#### 1.1 概念 ####

> 当需要在一段时间后执行某个任务时，可以使用 time.Timer。Timer 会在一段时间后向一个 channel 发送一个时间值，可以使用该 channel 来触发任务的执行。具体来说，当需要在一段时间后执行某个任务时，可以创建一个 Timer，然后使用 <-timer.C 语句从 Timer.C channel 中读取时间值，当读取到时间值时，就执行任务。

#### 1.2 用法 ####

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    func main() {
        
    	timer := time.NewTimer(2 * time.Second)
    	defer timer.Stop()
    	<-timer.C
    	fmt.Println("timer expired")
    }

结果：

timer expired

### 2 Ticker（定时执行多次） ###

#### 2.1 概念 ####

> > 当需要定期执行某个任务时，可以使用 time.Ticker。Ticker 会定期向一个 channel 发送一个时间值，可以使用该 channel 来触发定期任务的执行。具体来说，当需要定期执行某个任务时，可以创建一个 Ticker，然后使用 for range 循环从 Ticker.C channel 中读取时间值，每当读取到时间值时，就执行定期任务。

#### 2.2 用法 ####

package main
    
    import (
    	"fmt"
    	"time"
    )
    
    func main() {
        
    	ticker := time.NewTicker(1 * time.Second)
    	defer ticker.Stop()
    	for {
        
    		select {
        
    		case <-ticker.C:
    			fmt.Println("tick")
    		}
    	}
    }

结果：

tick
    tick
    tick
    tick
    ...

### 3 WaitGroup（等待goroutine全部完毕） ###

#### 3.1 概念 ####

> 当需要等待一组 goroutine 完成时，可以使用 sync.WaitGroup。WaitGroup 用于等待一组 goroutine 完成其工作。具体来说，当一个 goroutine 需要等待一组其它 goroutine 完成时，它可以调用 WaitGroup 的 Add 方法来增加计数器的值，然后启动这组 goroutine。每个 goroutine 完成时，都应该调用 Done 方法来减少计数器的值。最后，等待这组 goroutine 完成时，应该调用 Wait 方法，该方法会阻塞直到计数器的值为 0。

#### 3.2 用法 ####

基础用法：

package main
    
    import (
    	"fmt"
    	"sync"
    )
    
    func main() {
        
    	var wg sync.WaitGroup
    	for i := 0; i < 5; i++ {
        
    		wg.Add(1)
    		go func(i int) {
        
    			defer wg.Done()
    			fmt.Printf("goroutine %d\n", i)
    		}(i)
    	}
    	wg.Wait()
    	fmt.Println("all goroutines done")
    }

### 4 其他 ###

#### 4.1 golang实现生产者消费者模型 ####

##### ①简易版 #####

> 这个版本只包含一个生产者和一个消费者，使用goroutine和channel来实现消息传递。
> 
>  *  done变量用于在主线程中阻塞等待所有子协程的结束。在代码执行过程中，通过向 done 变量发送信号（done <- true）来表示一个子协程已完成任务。当所有子协程都完成任务后，执行 <-done 将阻塞主线程，等待所有子协程完成任务。而在 produce 和 consume 中，done 只用于简单的通信任务，不会阻塞主线程。

package main
    
    import (
    	"fmt"
    	_ "time"
    )
    
    func produce(ch chan<- int, done chan<- bool) {
        
    	for i := 1; i < 3; i++ {
        
    		ch <- i
    		fmt.Printf("生产者生产 %d\n", i)
    	}
    	done <- true
    }
    
    func consume(ch <-chan int) {
        
    	for i := range ch {
        
    		fmt.Printf("消费者消费 %d\n", i)
    	}
    }
    
    func main() {
        
    	ch := make(chan int)
    	done := make(chan bool)
    
    	go produce(ch, done)
    	go consume(ch)
    
    	<-done
    }

##### ②进阶版 #####

> 这个版本包含多个生产者和消费者，使用缓冲channel和waitgroup来实现。

type Task struct{
        }  //自己实际需要的数据结构
    producer()  //实际生产数据逻辑
    consumer()  //实际处理逻辑
    
    main()中的consumerNum(消费者个数)，
    channelLen(通道长度)也可根据实际需要修改

代码实现：

package main
    
    import (
    	"fmt"
    	"sync"
    )
    
    type Task struct {
        
    	Data string
    }
    
    var wg sync.WaitGroup
    
    //生产逻辑
    func producer(tasks chan Task) {
        
    	t := Task{
        }
    
    	for i := 62; i < 72; i++ {
        
    		t.Data = string(i)
    		tasks <- t
    	}
    }
    
    func producerDispatch(tasks chan Task) {
        
    	defer close(tasks)
    
    	producer(tasks)
    }
    
    //消费数据处理逻辑
    func consumer(task Task) {
        
    	fmt.Printf("consum task:%v\n", task)
    }
    
    func consumerDispatch(tasks chan Task) {
        
    	defer wg.Done()
    
    	for task := range tasks {
        
    		consumer(task)
    	}
    }
    
    func main() {
        
    	//消费者个数
    	var consumerNum = 10
    	var channelLen = 50
    
    	tasks := make(chan Task, channelLen)
    
        //当producer执行完成后关闭队列
    	go producerDispatch(tasks)
    
    	for i := 0; i < consumerNum; i++ {
        
    		wg.Add(1)
            //当consumer消费完后wg.Done()
    		go consumerDispatch(tasks)
    	}
        //等待全部goroutine完成
    	wg.Wait()
    	fmt.Println("all done")
    }

1.  生产者完成任务后调用close关闭通道
2.  因为通道关闭，所以消费者在取完所有任务后，会退出取任务的for循环。
3.  主线程main, 通过Wait()保证了在所有任务处理完后才退出。
4.  wg.Add必须放在main中。不然，有可能main执行完了，
5.  producerDispatch,consumerDispatch协程还没被调度到，这样就直接打印“all done”退出了。这也是在实际过程中踩的一个坑。

##### ③高级版 #####

> 这个版本包含多个生产者和消费者，使用无锁队列和多个协程池来实现。

package main
    
    import (
    	"fmt"
    	"math/rand"
    	"sync"
    	"time"
    )
    
    type Queue struct {
        
    	items []int
    	head  int
    	tail  int
    }
    
    func NewQueue(size int) *Queue {
        
    	return &Queue{
        make([]int, size), 0, 0}
    }
    
    func (q *Queue) Push(item int) bool {
        
    	next := (q.tail + 1) % len(q.items)
    	if next == q.head {
        
    		return false
    	}
    	q.items[q.tail] = item
    	q.tail = next
    	return true
    }
    
    func (q *Queue) Pop() (int, bool) {
        
    	if q.head == q.tail {
        
    		return 0, false
    	}
    	item := q.items[q.head]
    	q.head = (q.head + 1) % len(q.items)
    	return item, true
    }
    
    type WorkerPool struct {
        
    	workers []chan func()
    	wg      sync.WaitGroup
    }
    
    func NewWorkerPool(nWorker int) *WorkerPool {
        
    	pool := &WorkerPool{
        }
    	pool.workers = make([]chan func(), nWorker)
    	for i := range pool.workers {
        
    		pool.workers[i] = make(chan func())
    		pool.wg.Add(1)
    		go pool.workerLoop(pool.workers[i])
    	}
    	return pool
    }
    
    func (p *WorkerPool) workerLoop(worker chan func()) {
        
    	defer p.wg.Done()
    	for task := range worker {
        
    		task()
    	}
    }
    
    func (p *WorkerPool) AddTask(task func()) {
        
    	worker := p.workers[rand.Intn(len(p.workers))]
    	worker <- task
    }
    
    func (p *WorkerPool) Close() {
        
    	for i := range p.workers {
        
    		close(p.workers[i])
    	}
    	p.wg.Wait()
    }
    
    func produce(queue *Queue, pool *WorkerPool, pid int) {
        
    	for i := 1; i <= 3; i++ {
        
    		success := queue.Push(i * pid)
    		if !success {
        
    			fmt.Printf("生产者%d生产失败\n", pid)
    		} else {
        
    			fmt.Printf("生产者%d生产%d\n", pid, i*pid)
    			pool.AddTask(func() {
        
    				time.Sleep(time.Millisecond * time.Duration(rand.Intn(3000)))
    			})
    		}
    	}
    }
    
    func consume(queue *Queue, pool *WorkerPool, cid int) {
        
    	for {
        
    		item, ok := queue.Pop()
    		if !ok {
        
    			return
    		}
    		fmt.Printf("消费者%d消费%d\n", cid, item)
    		pool.AddTask(func() {
        
    			time.Sleep(time.Millisecond * time.Duration(rand.Intn(3000)))
    		})
    	}
    }
    
    func main() {
        
    	queue := NewQueue(5)
    	pool := NewWorkerPool(5)
    
    	nProducer := 3
    	for i := 1; i <= nProducer; i++ {
        
    		go produce(queue, pool, i)
    	}
    
    	nConsumer := 2
    	for i := 1; i <= nConsumer; i++ {
        
    		go consume(queue, pool, i)
    	}
    
    	time.Sleep(time.Second * 10)
    	pool.Close()
    }