避免在发送到通道时发生竞争条件

golang学习网今天将给大家带来《避免在发送到通道时发生竞争条件》，感兴趣的朋友请继续看下去吧！以下内容将会涉及到等等知识点，如果你是正在学习Golang或者已经是大佬级别了，都非常欢迎也希望大家都能给我建议评论哈~希望能帮助到大家！

go版本go1.11.2 darwin/amd64

我有以下代码示例，是为了演示目的而创建的：

package main

import (
    ...
)

type t struct {
    ctx context.context
    ch1 chan string
}

func new(ctx context.context) *t {
    t := &t{ctx: ctx}
    go t.run(2)
    return t

}

func (t *t) run(workers int) {
    t.ch1 = make(chan string)
    done := make(chan struct{})

    go func() {
        <-t.ctx.done()
        close(done)
        close(t.ch1)
    }()

    for i := 0; i < workers; i++ {
        go func() {
            for {
                select {
                case <-done:
                    return
                case m, ok := <-t.ch1:
                    if ok {
                        t.process(done, m)
                    }
                }
            }
        }()
    }
}

func (t *t) process(done <-chan struct{}, s string) {
    select {
    case <-done:
        return
    default:
        log.printf("processing %s", s)
        time.sleep(time.millisecond * 200)
    }
}

func (t *t) read() <-chan string {
    return t.ch1
}

func (t *t) write(s string) error {
    select {
    case <-t.ctx.done():
        return errors.new("consumer is closed today")
    case t.ch1 <- s:
        return nil
    }
}

func main() {
    ctx, cancel := context.withcancel(context.background())
    t := new(ctx)

    go func() {
        for m := range t.read() {
            log.printf("got %s", m)
        }
        <-ctx.done()
    }()

    for i := 0; i < 10; i++ {
        t.write(strconv.itoa(i))
    }
    cancel()
}

当我使用竞争检测器构建并运行它时，它会引发以下数据竞争：

go build -race ./test/ && ./test
==================
WARNING: DATA RACE
Read at 0x00c0000b6030 by goroutine 7:
  main.main.func1()
      /redacted/test/app.go:60 +0x42

Previous write at 0x00c0000b6030 by goroutine 6:
  main.(*T).run()
      /redacted/test/app.go:24 +0x6a

Goroutine 7 (running) created at:
  main.main()
      /redacted/test/app.go:76 +0xbc

Goroutine 6 (running) created at:
  main.New()
      /redacted/test/app.go:18 +0xcd
  main.main()
      /redacted/test/app.go:74 +0x86
==================
==================
WARNING: DATA RACE
Read at 0x00c0000b6030 by main goroutine:
  main.(*T).Write()
      /redacted/test/app.go:67 +0x8a
  main.main()
      /redacted/test/app.go:84 +0xdc

Previous write at 0x00c0000b6030 by goroutine 6:
  main.(*T).run()
      /redacted/test/app.go:24 +0x6a

Goroutine 6 (running) created at:
  main.New()
      /redacted/test/app.go:18 +0xcd
  main.main()
      /redacted/test/app.go:74 +0x86
==================
2019/01/20 10:48:51 got 0
2019/01/20 10:48:51 got 3
2019/01/20 10:48:51 processing 1
2019/01/20 10:48:51 processing 2
2019/01/20 10:48:51 got 4
2019/01/20 10:48:51 got 5
2019/01/20 10:48:51 got 6
2019/01/20 10:48:51 got 7
2019/01/20 10:48:51 got 8
2019/01/20 10:48:51 got 9
Found 2 data race(s)

我遇到的问题是，我似乎无法找到一种方法让用户在不暴露任何写入通道的情况下向通道中输入某些内容，而无需竞争。这怎么可能呢？我是否缺少更好的模式？

解决方案

我建议进行以下更改：

• 在 new 中赋值给 ch1 以避免多个 goroutine 中对 t.ch1 的读写竞争
• 仅在对 write 的所有调用完成后才关闭 ch1，以避免“在关闭的通道上发送”恐慌
• 使用 sync.waitgroup 在写入所有值后等待所有处理 goroutine 完成（以便程序在处理完成之前不会退出）

将这些更改放在一起，如下所示：

package main

import (
    "log"
    "strconv"
    "sync"
    "time"
)

type T struct {
    // ch1 receives the values to process
    ch1 chan string

    // wg is used to wait for the workers to stop
    wg sync.WaitGroup
}

func New() *T {
    t := &T{
        ch1: make(chan string),
    }
    go t.run(2)
    return t
}

func (t *T) run(workers int) {
    // add the workers to the WaitGroup
    t.wg.Add(workers)

    for i := 0; i < workers; i++ {
        go func() {
            // process values from the channel until it closes
            // and then signal to the WaitGroup that we're done
            defer t.wg.Done()
            for m := range t.ch1 {
                t.process(m)
            }
        }()
    }
}

// Stop is called after we're done calling Write and we want to stop the
// processing once all values have been processed
func (t *T) Stop() {
    // close t.ch1 so that the workers know to stop processing
    close(t.ch1)

    // wait for the workers to all finish before returning
    t.wg.Wait()
}

func (t *T) process(s string) {
    log.Printf("processing %s", s)
    time.Sleep(time.Millisecond * 200)
}

func (t *T) Write(s string) {
    t.ch1 <- s
}

func main() {
    // start the main loop
    t := New()

    // write 10 values
    for i := 0; i < 10; i++ {
        t.Write(strconv.Itoa(i))
    }

    // stop the loop, which will wait for processing to finish before returning
    t.Stop()
}

今天带大家了解了的相关知识，希望对你有所帮助；关于Golang的技术知识我们会一点点深入介绍，欢迎大家关注golang学习网公众号，一起学习编程~