• Go 之常用并发学习

    在 Go语言实战 中看到有些并发相关的例子，讲解得也比较详细，于是乎写来加深下印象。

    无缓冲通道

    无缓冲通道在接收前没有能力保存任何值。我自己找了书上的示例来加深一下印象。

    模拟网球比赛

    1. package main
    2.  
    3. import (
    4. 	"fmt"
    5. 	"math/rand"
    6. 	"sync"
    7. 	"time"
    8. )
    9.  
    10. var (
    11. 	wg sync.WaitGroup
    12. )
    13.  
    14. func init() {
    15. 	rand.Seed(time.Now().UnixNano()) // 让每次运行生成的随机数不相同
    16. }
    17.  
    18. func main() {
    19. 	count := make(chan int)
    20. 	wg.Add(2)
    21. 	go player("Nadal", count)
    22. 	go player("Looking", count)
    23.  
    24. 	count <- 1
    25. 	wg.Wait()
    26. }
    27.  
    28. func player(name string, count chan int) {
    29. 	defer wg.Done()
    30.  
    31. 	for {
    32. 		ball, ok := <-count
    33. 		if !ok {
    34. 			fmt.Printf("Player %s Win\n", name)
    35. 			return
    36. 		}
    37. 		n := rand.Intn(100)
    38. 		if n%13 == 0 {
    39. 			fmt.Printf("Player %s Missed\n", name)
    40. 			close(count)
    41. 			return
    42. 		}
    43. 		fmt.Printf("Player %s Hit %d\n", name, ball)
    44. 		count <- ball
    45. 	}
    46. }

    模拟接力赛

    接力赛中，接力棒只能在一个人手中。

    1. package main
    2.  
    3. import (
    4. 	"fmt"
    5. 	"sync"
    6. 	"time"
    7. )
    8.  
    9. var (
    10. 	wg sync.WaitGroup
    11. )
    12.  
    13.  
    14. func main() {
    15. 	baton := make(chan int)
    16. 	wg.Add(1)
    17. 	go Runner(baton)
    18. 	baton <- 1
    19. 	wg.Wait()
    20. }
    21.  
    22. func Runner(baton chan int)  {
    23. 	var newRunner int
    24. 	runner := <- baton
    25. 	fmt.Printf("Runner %d Running with baton\n", runner)
    26. 	if runner != 4 {
    27. 		newRunner = runner + 1
    28. 		fmt.Printf("Runner %d to the line\n", newRunner)
    29. 		go Runner(baton)
    30. 	}
    31. 	time.Sleep(100 * time.Millisecond)
    32. 	if runner == 4 {
    33. 		fmt.Printf("Runner %d finished, Race over\n", runner)
    34. 		wg.Done()
    35. 		return
    36. 	}
    37. 	fmt.Printf("Runner %d Exchange with runner %d \n", runner, newRunner)
    38. 	baton <- newRunner
    39. }
    1. Runner 1 Running with baton
    2. Runner 2 to the line
    3. Runner 1 Exchange with runner 2 
    4. Runner 2 Running with baton
    5. Runner 3 to the line
    6. Runner 2 Exchange with runner 3 
    7. Runner 3 Running with baton
    8. Runner 4 to the line
    9. Runner 3 Exchange with runner 4 
    10. Runner 4 Running with baton
    11. Runner 4 finished, Race over

    有缓冲通道

    缓冲通道不强制发送和接收同时完成。

    当通道关闭后，goroutine 依旧可以从通道接收数据，但是不能再向通道里发送数据。能够从已经关闭的通道接收数据这一点非常重要，因为这允许通 道关闭后依旧能取出其中缓冲的全部值，而不会有数据丢失。从一个已经关闭且没有数据的通道 里获取数据，总会立刻返回，并返回一个通道类型的零值。如果在获取通道时还加入了可选的标 志，就能得到通道的状态信息。

    模拟任务分发和处理

    1. package main
    2.  
    3. import (
    4. 	"fmt"
    5. 	"math/rand"
    6. 	"sync"
    7. 	"time"
    8. )
    9.  
    10. const (
    11. 	numberGoroutines = 4
    12. 	taskLoad         = 10
    13. )
    14.  
    15. var wg sync.WaitGroup
    16.  
    17. func init() {
    18. 	rand.Seed(time.Now().UnixNano())
    19. }
    20.  
    21. func main() {
    22. 	tasks := make(chan string, taskLoad)
    23. 	wg.Add(numberGoroutines)
    24. 	for gr := 1; gr <= numberGoroutines; gr++ {
    25. 		go worker(tasks, gr)
    26. 	}
    27. 	for post := 1; post <= taskLoad; post++ {
    28. 		tasks <- fmt.Sprintf("Task: %d", post)
    29. 	}
    30. 	close(tasks) // 任务发布完毕后关闭通道，关闭通道不影响其它Goroutine对已发布内容的正常接收
    31. 	wg.Wait()
    32. }
    33.  
    34. func worker(tasks chan string, worker int) {
    35. 	defer wg.Done()
    36.  
    37. 	for {
    38. 		task, ok := <-tasks
    39. 		if !ok {
    40. 			fmt.Printf("Worker: %d : Shutting down\n", worker)
    41. 			return
    42. 		}
    43. 		fmt.Printf("Worker: %d : Started %s\n", worker, task)
    44. 		sleep := rand.Int63n(100)
    45. 		time.Sleep(time.Duration(sleep) * time.Millisecond)
    46. 		fmt.Printf("Worker: %d : Completed %s\n", worker, task)
    47. 	}
    48. }
    1. Worker: 4 : Started Task: 4
    2. Worker: 2 : Started Task: 2
    3. Worker: 3 : Started Task: 3
    4. Worker: 1 : Started Task: 1
    5. Worker: 3 : Completed Task: 3
    6. Worker: 3 : Started Task: 5
    7. Worker: 2 : Completed Task: 2
    8. Worker: 2 : Started Task: 6
    9. Worker: 1 : Completed Task: 1
    10. Worker: 1 : Started Task: 7
    11. Worker: 1 : Completed Task: 7
    12. Worker: 1 : Started Task: 8
    13. Worker: 3 : Completed Task: 5
    14. Worker: 3 : Started Task: 9
    15. Worker: 4 : Completed Task: 4
    16. Worker: 4 : Started Task: 10
    17. Worker: 2 : Completed Task: 6
    18. Worker: 2 : Shutting down
    19. Worker: 3 : Completed Task: 9
    20. Worker: 3 : Shutting down
    21. Worker: 1 : Completed Task: 8
    22. Worker: 1 : Shutting down
    23. Worker: 4 : Completed Task: 10
    24. Worker: 4 : Shutting down

    runner

    runner 包用于展示如何使用通道来监视程序的执行时间，如果程序运行时间太长，也可以
    用runner 包来终止程序。当开发需要调度后台处理任务的程序的时候，这种模式会很有用。

    runner/runner.go

    1. package runner
    2.  
    3. import (
    4. 	"errors"
    5. 	"os"
    6. 	"os/signal"
    7. 	"time"
    8. )
    9.  
    10. type Runner struct {
    11. 	interrupt chan os.Signal
    12. 	complete  chan error
    13. 	timeout   <-chan time.Time // 单向通道，只允许接收
    14. 	tasks     []func(int)
    15. }
    16.  
    17. var ErrTimeout = errors.New("received timeout")
    18. var ErrInterrupt = errors.New("received interrupt")
    19.  
    20. func New(d time.Duration) *Runner {
    21. 	return &Runner{
    22. 		interrupt: make(chan os.Signal, 1),
    23. 		complete:  make(chan error),
    24. 		timeout:   time.After(d),
    25. 	}
    26. }
    27.  
    28. func (r *Runner) Add(tasks ...func(int)) {
    29. 	r.tasks = append(r.tasks, tasks...)
    30. }
    31.  
    32. func (r *Runner) Start() error {
    33. 	signal.Notify(r.interrupt, os.Interrupt) // 如果有中断，会将中断信号发送到 r.interrupt
    34. 	go func() {
    35. 		r.complete <- r.run()
    36. 	}()
    37. 	select {
    38. 	case err := <-r.complete:
    39. 		return err // 如果提前中断，err 是 ErrInterrupt，正常结束则是 nil
    40. 	case <-r.timeout:
    41. 		return ErrTimeout
    42. 	}
    43. }
    44.  
    45. func (r *Runner) run() error {
    46. 	for id, task := range r.tasks {
    47. 		if r.gotInterrupt() {
    48. 			return ErrInterrupt
    49. 		}
    50. 		task(id)
    51. 	}
    52. 	return nil
    53. }
    54.  
    55. func (r *Runner) gotInterrupt() bool {
    56. 	select {
    57. 	case <-r.interrupt:
    58. 		signal.Stop(r.interrupt)
    59. 		return true
    60. 	default:
    61. 		return false
    62. 	}
    63. }

    main.go 

    1. package main
    2.  
    3. import (
    4. 	"github.com/test/runner"
    5. 	"log"
    6. 	"os"
    7. 	"time"
    8. )
    9.  
    10. const timeout = 3 * time.Second
    11.  
    12. func main() {
    13. 	log.Println("Starting work.")
    14. 	r := runner.New(timeout)
    15. 	r.Add(createTask(), createTask(), createTask())
    16.  
    17. 	if err := r.Start(); err != nil {
    18. 		switch err {
    19. 		case runner.ErrTimeout:
    20. 			log.Println("Terminating due to timeout.")
    21. 			os.Exit(1)
    22. 		case runner.ErrInterrupt:
    23. 			log.Println("Terminating due to interrupt.")
    24. 			os.Exit(2)
    25. 		}
    26. 	}
    27. 	log.Println("Process ended.")
    28. }
    29.  
    30. func createTask() func(int) {
    31. 	return func(id int) {
    32. 		log.Printf("Processor - Task #%d.", id)
    33. 		time.Sleep(time.Duration(id) * time.Second)
    34. 	}
    35. }

    超时退出

    1. 2024/04/21 17:26:30 Starting work.
    2. 2024/04/21 17:26:30 Processor - Task #0.
    3. 2024/04/21 17:26:30 Processor - Task #1.
    4. 2024/04/21 17:26:31 Processor - Task #2.
    5. 2024/04/21 17:26:33 Terminating due to timeout.

    中断退出

    1. 2024/04/21 17:28:18 Starting work.
    2. 2024/04/21 17:28:18 Processor - Task #0.
    3. 2024/04/21 17:28:18 Processor - Task #1.
    4. 2024/04/21 17:28:19 Terminating due to interrupt.

    正常退出

    1. 2024/04/21 17:30:40 Starting work.
    2. 2024/04/21 17:30:40 Processor - Task #0.
    3. 2024/04/21 17:30:40 Processor - Task #1.
    4. 2024/04/21 17:30:41 Processor - Task #2.
    5. 2024/04/21 17:30:43 Process ended.

    pool

    pool 使用有缓冲通道实现资源池。

    pool/pool.go

    1. package pool
    2.  
    3. import (
    4. 	"errors"
    5. 	"io"
    6. 	"log"
    7. 	"sync"
    8. )
    9.  
    10. type Pool struct {
    11. 	m         sync.Mutex
    12. 	resources chan io.Closer
    13. 	factory   func() (io.Closer, error)
    14. 	closed    bool
    15. }
    16.  
    17. var ErrPoolClosed = errors.New("pool has been closed")
    18.  
    19. func New(fn func() (io.Closer, error), size uint) (*Pool, error) {
    20. 	if size <= 0 {
    21. 		return nil, errors.New("size value too small")
    22. 	}
    23. 	return &Pool{
    24. 		resources: make(chan io.Closer, size),
    25. 		factory:   fn,
    26. 	}, nil
    27. }
    28.  
    29. func (p *Pool) Acquire() (io.Closer, error) {
    30. 	select {
    31. 	case r, ok := <-p.resources:
    32. 		log.Println("Acquire:", "Shared resource")
    33. 		if !ok {
    34. 			return nil, ErrPoolClosed
    35. 		}
    36. 		return r, nil
    37. 	default:
    38. 		log.Println("Acquire:", "New resource")
    39. 		return p.factory()
    40. 	}
    41. }
    42.  
    43. func (p *Pool) Release(r io.Closer) {
    44. 	p.m.Lock()
    45. 	defer p.m.Unlock()
    46. 	if p.closed {
    47. 		r.Close()
    48. 		return
    49. 	}
    50. 	select {
    51. 	case p.resources <- r:
    52. 		log.Println("Release:", "In queue")
    53. 	default:
    54. 		log.Println("Release:", "Closing")
    55. 		r.Close()
    56. 	}
    57. }
    58.  
    59. func (p *Pool) Close() {
    60. 	p.m.Lock()
    61. 	defer p.m.Unlock()
    62. 	if p.closed {
    63. 		return
    64. 	}
    65. 	p.closed = true
    66. 	close(p.resources)
    67. 	for r := range p.resources {
    68. 		r.Close()
    69. 	}
    70. }

    main.go

    1. package main
    2.  
    3. import (
    4. 	"github.com/test/pool"
    5. 	"io"
    6. 	"log"
    7. 	"math/rand"
    8. 	"sync"
    9. 	"sync/atomic"
    10. 	"time"
    11. )
    12.  
    13. const (
    14. 	maxGoroutines   = 5
    15. 	pooledResources = 2
    16. )
    17.  
    18. type dbConnection struct {
    19. 	ID int32
    20. }
    21.  
    22. func (dbConn *dbConnection) Close() error {
    23. 	log.Println("Close: Connection", dbConn.ID)
    24. 	return nil
    25. }
    26.  
    27. var idCounter int32
    28.  
    29. func createConnection() (io.Closer, error) {
    30. 	id := atomic.AddInt32(&idCounter, 1)
    31. 	log.Println("Create: New Connection", id)
    32.  
    33. 	return &dbConnection{id}, nil
    34. }
    35.  
    36. func main() {
    37. 	var wg sync.WaitGroup
    38. 	wg.Add(maxGoroutines)
    39.  
    40. 	p, err := pool.New(createConnection, pooledResources)
    41. 	if err != nil {
    42. 		log.Println(err)
    43. 	}
    44.  
    45. 	for query := 0; query < maxGoroutines; query++ {
    46. 		go func(q int) {
    47. 			defer wg.Done()
    48. 			performQueries(q, p)
    49. 		}(query)
    50. 	}
    51. 	wg.Wait()
    52. 	log.Println("Shutdown program")
    53. 	p.Close()
    54. }
    55.  
    56. func performQueries(query int, p *pool.Pool) {
    57. 	conn, err := p.Acquire()
    58. 	if err != nil {
    59. 		log.Println(err)
    60. 		return
    61. 	}
    62. 	defer p.Release(conn)
    63. 	time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond)
    64. 	log.Printf("QID[%d] CID[%d]\n", query, conn.(*dbConnection).ID)
    65. }

    运行结果 

    1. 2024/04/21 20:29:20 Acquire: New resource
    2. 2024/04/21 20:29:20 Create: New Connection 1
    3. 2024/04/21 20:29:20 Acquire: New resource
    4. 2024/04/21 20:29:20 Acquire: New resource
    5. 2024/04/21 20:29:20 Create: New Connection 3
    6. 2024/04/21 20:29:20 Acquire: New resource
    7. 2024/04/21 20:29:20 Create: New Connection 4
    8. 2024/04/21 20:29:20 Create: New Connection 2
    9. 2024/04/21 20:29:20 Acquire: New resource
    10. 2024/04/21 20:29:20 Create: New Connection 5
    11. 2024/04/21 20:29:20 QID[0] CID[2]
    12. 2024/04/21 20:29:20 Release: In queue
    13. 2024/04/21 20:29:20 QID[2] CID[5]
    14. 2024/04/21 20:29:20 Release: In queue
    15. 2024/04/21 20:29:20 QID[4] CID[1]
    16. 2024/04/21 20:29:20 Release: Closing
    17. 2024/04/21 20:29:20 Close: Connection 1
    18. 2024/04/21 20:29:21 QID[1] CID[4]
    19. 2024/04/21 20:29:21 Release: Closing
    20. 2024/04/21 20:29:21 Close: Connection 4
    21. 2024/04/21 20:29:21 QID[3] CID[3]
    22. 2024/04/21 20:29:21 Release: Closing
    23. 2024/04/21 20:29:21 Close: Connection 3
    24. 2024/04/21 20:29:21 Shutdown program
    25. 2024/04/21 20:29:21 Close: Connection 2
    26. 2024/04/21 20:29:21 Close: Connection 5

    work

    work 使用无缓冲通道创建资源池。work 新建时就生成指定个数 goroutine 循环等待（无缓冲通道无数据阻塞）消费任务，然后主线程再分发相应任务到通道，消费 goroutine 再继续执行消费任务。

    work/work.go

    1. package work
    2.  
    3. import "sync"
    4.  
    5. type Worker interface {
    6. 	Task()
    7. }
    8.  
    9. type Pool struct {
    10. 	work chan Worker
    11. 	wg   sync.WaitGroup
    12. }
    13.  
    14. func New(maxGoroutines int) *Pool {
    15. 	p := Pool{
    16. 		work: make(chan Worker),
    17. 	}
    18. 	p.wg.Add(maxGoroutines)
    19. 	for i := 0; i < maxGoroutines; i++ {
    20. 		go func() {
    21. 			defer p.wg.Done()
    22. 			for w := range p.work { // 如果 work 关闭，for range 循环结束
    23. 				w.Task()
    24. 			}
    25. 		}()
    26. 	}
    27. 	return &p
    28. }
    29.  
    30. func (p *Pool) Run(w Worker) {
    31. 	p.work <- w
    32. }
    33.  
    34. func (p *Pool) Shutdown() {
    35. 	close(p.work) // 关闭通道，避免 New 产生的 goroutine 一直阻塞不退出
    36. 	p.wg.Wait()
    37. }

    main.go

    1. package main
    2.  
    3. import (
    4. 	"github.com/test/work"
    5. 	"log"
    6. 	"sync"
    7. 	"time"
    8. )
    9.  
    10. var names = []string{
    11. 	"steve",
    12. 	"jason",
    13. 	"looking",
    14. }
    15.  
    16. type namePrinter struct {
    17. 	name string
    18. }
    19.  
    20. func (m *namePrinter) Task() {
    21. 	log.Println(m.name)
    22. 	time.Sleep(time.Second)
    23. }
    24.  
    25. const times = 3
    26.  
    27. func main() {
    28. 	p := work.New(2)
    29.  
    30. 	var wg sync.WaitGroup
    31. 	wg.Add(times * len(names))
    32. 	for i := 0; i < times; i++ {
    33. 		for _, name := range names {
    34. 			np := namePrinter{name: name}
    35. 			go func() {
    36. 				defer wg.Done()
    37. 				p.Run(&np)
    38. 			}()
    39. 		}
    40. 	}
    41. 	wg.Wait()
    42. 	p.Shutdown()
    43. }

    运行结果

    1. 2024/04/21 21:39:28 steve
    2. 2024/04/21 21:39:28 looking
    3. 2024/04/21 21:39:29 jason
    4. 2024/04/21 21:39:29 looking
    5. 2024/04/21 21:39:30 steve
    6. 2024/04/21 21:39:30 jason
    7. 2024/04/21 21:39:31 looking
    8. 2024/04/21 21:39:31 steve
    9. 2024/04/21 21:39:32 jason
  • 相关阅读:
    XML序列化与反序列化接口对接实战，看这篇就够了
    Python Http请求和HTML的解析
    立哥国家示范项目-5G智慧文旅
    什么是war包？war包该怎么运行？
    线上频繁fullgc问题-SpringActuator的坑
    全网最简约的Vscode配置Anaconda环境（百分百成功）
    Vue3引入pinia并模块化
    Java 去除字符中的空格、回车符
    ArangoDB 学习笔记（一）
    数据库索引原理
  • 原文地址：https://blog.csdn.net/TomorrowAndTuture/article/details/138031256