如何在 goroutine 之间共享地图

【问题标题】：How to share a map across goroutines如何在 goroutine 之间共享地图
【发布时间】：2018-09-13 16:56:32
【问题描述】：

我正在尝试在 Go 中编写一个通知结构，它将包含一系列键及其各自的值，如果值低于阈值，则会触发通知。

通知应该只触发一次，当第一个样本低于阈值时，低于该阈值的其他样本不应再次触发，直到值上升到阈值以上。

例如，假设我的阈值为 10，我发送 15、14、11、10、... 9 的样本。一旦发送 9，就应该触发通知。 8、7、4 的进一步样品不应造成任何影响。以下样本如 5、6、7、9、10、11、14、30 应该什么都不做。一旦样本再次低于 10: 30, 20, 15, 10, 7... 必须发送另一个通知。

当多个 goroutine 操作我的结构时，我遇到了问题。

我尝试使用 sync.Mutex 进行同步，并且还使用了 sync.Map，但没有运气。我感觉某处有参考副本或缓存，但我对 Go 太陌生了，无法找到问题。

为此，我创建了一个这样的结构：

type Notifier interface {
    Send(message string)
}

type NotificationBoard struct {
    mutex    sync.Mutex
    Last     sync.Map
    notifier Notifier
}

func (n *NotificationBoard) Init(notifier Notifier) {
    n.notifier = notifier
}

// NotifyLess ...
func (n *NotificationBoard) NotifyLess(key string, value, threshold float64) {
    n.mutex.Lock()
    defer n.mutex.Unlock()

    if value >= threshold {
        fmt.Printf("NotificationBoard.NotifyLess %v (value >= threshold): %v >= %v\n", key, value, threshold)
        n.Last.Store(key, value)
        return
    }

    // value < threshold
    if last, found := n.Last.Load(key); found == true {
        fmt.Printf("NotificationBoard.NotifyLess %v (value < threshold): %v < %v : found %v\n", key, value, threshold, last)
        if last.(float64) >= threshold { // first trigger
            n.notifier.Send(fmt.Sprintf("%s < %v (%v)", key, threshold, value))
        }
    } else {
        fmt.Printf("NotificationBoard.NotifyLess %v (value < threshold): %v < %v : not found\n", key, value, threshold)
        // not found, started board as less
        n.notifier.Send(fmt.Sprintf("%s < %v (%v)", key, threshold, value))
    }

    n.Last.Store(key, value)
    return
}

我知道使用 sync.Mutex 或 sync.Map 就足够了，但上面的代码两者都有，因为它是我当前的（损坏的）版本。

为了测试，我设置了以下代码：

type dummy struct{}

func (d *dummy) Send(message string) {
    fmt.Println("--------------> notifying", message)
}

func newBoard() *NotificationBoard {
    notificationBoard := &NotificationBoard{}
    notificationBoard.Init(&dummy{})
    return notificationBoard
}

我还添加了一些 fmt.Println 跟踪（为简洁起见，未包含在我的代码中）并首先准备了一个单 gouroutine 测试（按预期工作）：

func Test1(t *testing.T) {
    board := newBoard()
    board.NotifyLess("k1", 15, 10)
    board.NotifyLess("k1", 10, 10)
    board.NotifyLess("k1", 5, 10)
    board.NotifyLess("k1", 4, 10)
    board.NotifyLess("k1", 3, 10)
    board.NotifyLess("k1", 10, 10)
    board.NotifyLess("k1", 15, 10)
    board.NotifyLess("k1", 20, 10)
    board.NotifyLess("k1", 15, 10)
    board.NotifyLess("k1", 10, 10)
    board.NotifyLess("k1", 5, 10)
    board.NotifyLess("k1", 1, 10)
}

输出：

> go test -run Test1
NotificationBoard.NotifyLess k1 (value >= threshold): 15 >= 10
NotificationBoard.NotifyLess k1 (value >= threshold): 10 >= 10
NotificationBoard.NotifyLess k1 (value < threshold): 5 < 10 : found 10
--------------> notifying k1 < 10 (5)
NotificationBoard.NotifyLess k1 (value < threshold): 4 < 10 : found 5
NotificationBoard.NotifyLess k1 (value < threshold): 3 < 10 : found 4
NotificationBoard.NotifyLess k1 (value >= threshold): 10 >= 10
NotificationBoard.NotifyLess k1 (value >= threshold): 15 >= 10
NotificationBoard.NotifyLess k1 (value >= threshold): 20 >= 10
NotificationBoard.NotifyLess k1 (value >= threshold): 15 >= 10
NotificationBoard.NotifyLess k1 (value >= threshold): 10 >= 10
NotificationBoard.NotifyLess k1 (value < threshold): 5 < 10 : found 10
--------------> notifying k1 < 10 (5)
NotificationBoard.NotifyLess k1 (value < threshold): 1 < 10 : found 5
PASS

我们可以看到输出“notifying ....”发生了两次，只是在样本低于阈值的那一刻

但是，我创建了一个多 gouroutine 测试，然后通知发生了多次：

func Test3(t *testing.T) {
    preparing := sync.WaitGroup{}
    preparing.Add(1)

    board := newBoard()
    wg := sync.WaitGroup{}

    for i := 0; i < 30; i++ {
        wg.Add(1)
        go func(x int, not *NotificationBoard) {
            fmt.Printf("routine %v waiting preparation... \n", x)
            preparing.Wait()

            for j := 15.0; j > 5; j-- {
                fmt.Printf("routine %v notifying %v\n", x, j)
                not.NotifyLess("keyX", j+float64(x+1)/100, 10)
            }

            wg.Done()
        }(i, board)
    }

    preparing.Done()
    wg.Wait()

}

哪个输出：

> go test -run Test3
routine 7 waiting preparation...
routine 2 waiting preparation...
routine 2 notifying 15
NotificationBoard.NotifyLess keyX (value >= threshold): 15.03 >= 10
routine 2 notifying 14
NotificationBoard.NotifyLess keyX (value >= threshold): 14.03 >= 10
routine 2 notifying 13
NotificationBoard.NotifyLess keyX (value >= threshold): 13.03 >= 10
routine 2 notifying 12
NotificationBoard.NotifyLess keyX (value >= threshold): 12.03 >= 10
routine 2 notifying 11
NotificationBoard.NotifyLess keyX (value >= threshold): 11.03 >= 10
routine 2 notifying 10
NotificationBoard.NotifyLess keyX (value >= threshold): 10.03 >= 10
routine 2 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.03 < 10 : found 10.03
--------------> notifying keyX < 10 (9.03)
routine 2 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 8.03 < 10 : found 9.03
routine 2 notifying 7
NotificationBoard.NotifyLess keyX (value < threshold): 7.03 < 10 : found 8.03
routine 2 notifying 6
NotificationBoard.NotifyLess keyX (value < threshold): 6.03 < 10 : found 7.03
routine 14 waiting preparation...
routine 14 notifying 15
NotificationBoard.NotifyLess keyX (value >= threshold): 15.15 >= 10
routine 14 notifying 14
NotificationBoard.NotifyLess keyX (value >= threshold): 14.15 >= 10
routine 14 notifying 13
NotificationBoard.NotifyLess keyX (value >= threshold): 13.15 >= 10
routine 14 notifying 12
NotificationBoard.NotifyLess keyX (value >= threshold): 12.15 >= 10
routine 14 notifying 11
NotificationBoard.NotifyLess keyX (value >= threshold): 11.15 >= 10
routine 14 notifying 10
NotificationBoard.NotifyLess keyX (value >= threshold): 10.15 >= 10
routine 14 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.15 < 10 : found 10.15
--------------> notifying keyX < 10 (9.15)
routine 14 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 8.15 < 10 : found 9.15
routine 14 notifying 7
NotificationBoard.NotifyLess keyX (value < threshold): 7.15 < 10 : found 8.15
routine 14 notifying 6
NotificationBoard.NotifyLess keyX (value < threshold): 6.15 < 10 : found 7.15
routine 22 waiting preparation...
routine 27 waiting preparation...
routine 27 notifying 15
NotificationBoard.NotifyLess keyX (value >= threshold): 15.28 >= 10
routine 27 notifying 14
NotificationBoard.NotifyLess keyX (value >= threshold): 14.28 >= 10
routine 27 notifying 13
NotificationBoard.NotifyLess keyX (value >= threshold): 13.28 >= 10
routine 27 notifying 12
NotificationBoard.NotifyLess keyX (value >= threshold): 12.28 >= 10
routine 27 notifying 11
NotificationBoard.NotifyLess keyX (value >= threshold): 11.28 >= 10
routine 27 notifying 10
NotificationBoard.NotifyLess keyX (value >= threshold): 10.28 >= 10
routine 27 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.28 < 10 : found 10.28
--------------> notifying keyX < 10 (9.28)
routine 27 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 8.28 < 10 : found 9.28
routine 27 notifying 7
NotificationBoard.NotifyLess keyX (value < threshold): 7.28 < 10 : found 8.28
routine 27 notifying 6
NotificationBoard.NotifyLess keyX (value < threshold): 6.28 < 10 : found 7.28
routine 20 waiting preparation...
routine 20 notifying 15
NotificationBoard.NotifyLess keyX (value >= threshold): 15.21 >= 10
routine 20 notifying 14
NotificationBoard.NotifyLess keyX (value >= threshold): 14.21 >= 10
routine 20 notifying 13
NotificationBoard.NotifyLess keyX (value >= threshold): 13.21 >= 10
routine 20 notifying 12
NotificationBoard.NotifyLess keyX (value >= threshold): 12.21 >= 10
routine 20 notifying 11
NotificationBoard.NotifyLess keyX (value >= threshold): 11.21 >= 10
routine 20 notifying 10
NotificationBoard.NotifyLess keyX (value >= threshold): 10.21 >= 10
routine 20 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.21 < 10 : found 10.21
--------------> notifying keyX < 10 (9.21)
routine 20 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 8.21 < 10 : found 9.21
routine 20 notifying 7
NotificationBoard.NotifyLess keyX (value < threshold): 7.21 < 10 : found 8.21
routine 20 notifying 6
NotificationBoard.NotifyLess keyX (value < threshold): 6.21 < 10 : found 7.21
routine 19 waiting preparation...
routine 19 notifying 15
NotificationBoard.NotifyLess keyX (value >= threshold): 15.2 >= 10
routine 19 notifying 14
NotificationBoard.NotifyLess keyX (value >= threshold): 14.2 >= 10
routine 19 notifying 13
NotificationBoard.NotifyLess keyX (value >= threshold): 13.2 >= 10
routine 19 notifying 12
NotificationBoard.NotifyLess keyX (value >= threshold): 12.2 >= 10
routine 19 notifying 11
NotificationBoard.NotifyLess keyX (value >= threshold): 11.2 >= 10
routine 19 notifying 10
NotificationBoard.NotifyLess keyX (value >= threshold): 10.2 >= 10
routine 19 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.2 < 10 : found 10.2
--------------> notifying keyX < 10 (9.2)
routine 19 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 8.2 < 10 : found 9.2
routine 19 notifying 7
NotificationBoard.NotifyLess keyX (value < threshold): 7.2 < 10 : found 8.2
routine 19 notifying 6
NotificationBoard.NotifyLess keyX (value < threshold): 6.2 < 10 : found 7.2
routine 0 waiting preparation...
routine 0 notifying 15
NotificationBoard.NotifyLess keyX (value >= threshold): 15.01 >= 10
routine 0 notifying 14
NotificationBoard.NotifyLess keyX (value >= threshold): 14.01 >= 10
routine 0 notifying 13
NotificationBoard.NotifyLess keyX (value >= threshold): 13.01 >= 10
routine 0 notifying 12
NotificationBoard.NotifyLess keyX (value >= threshold): 12.01 >= 10
routine 0 notifying 11
NotificationBoard.NotifyLess keyX (value >= threshold): 11.01 >= 10
routine 0 notifying 10
NotificationBoard.NotifyLess keyX (value >= threshold): 10.01 >= 10
routine 0 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.01 < 10 : found 10.01
--------------> notifying keyX < 10 (9.01)
routine 0 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 8.01 < 10 : found 9.01
routine 0 notifying 7
NotificationBoard.NotifyLess keyX (value < threshold): 7.01 < 10 : found 8.01
routine 0 notifying 6
NotificationBoard.NotifyLess keyX (value < threshold): 6.01 < 10 : found 7.01
routine 17 waiting preparation...
routine 17 notifying 15
NotificationBoard.NotifyLess keyX (value >= threshold): 15.18 >= 10
routine 17 notifying 14
NotificationBoard.NotifyLess keyX (value >= threshold): 14.18 >= 10
routine 17 notifying 13
NotificationBoard.NotifyLess keyX (value >= threshold): 13.18 >= 10
routine 17 notifying 12
NotificationBoard.NotifyLess keyX (value >= threshold): 12.18 >= 10
routine 17 notifying 11
NotificationBoard.NotifyLess keyX (value >= threshold): 11.18 >= 10
routine 17 notifying 10
NotificationBoard.NotifyLess keyX (value >= threshold): 10.18 >= 10
routine 17 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.18 < 10 : found 10.18
--------------> notifying keyX < 10 (9.18)
routine 17 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 8.18 < 10 : found 9.18
routine 17 notifying 7
NotificationBoard.NotifyLess keyX (value < threshold): 7.18 < 10 : found 8.18
routine 17 notifying 6
NotificationBoard.NotifyLess keyX (value < threshold): 6.18 < 10 : found 7.18
routine 15 waiting preparation...
routine 16 waiting preparation...
...... continues

我添加了一个十进制值来表示 goroutine，然后查看输出，似乎每个 goroutine 都有自己的映射副本，因为它们找到了具有相同小数位的先前值。但后来我发现：

...
NotificationBoard.NotifyLess keyX (value >= threshold): 10.22 >= 10
routine 21 notifying 9
NotificationBoard.NotifyLess keyX (value >= threshold): 10.07 >= 10
routine 6 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.08 < 10 : found 10.07
--------------> notifying keyX < 10 (9.08)
routine 7 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 9.17 < 10 : found 9.08
routine 16 notifying 8
NotificationBoard.NotifyLess keyX (value >= threshold): 10.11 >= 10
routine 10 notifying 9
NotificationBoard.NotifyLess keyX (value < threshold): 9.3 < 10 : found 10.11
--------------> notifying keyX < 10 (9.3)
routine 29 notifying 8
NotificationBoard.NotifyLess keyX (value < threshold): 9.19 < 10 : found 9.3
routine 18 notifying 8
...

这表明他们也在从其他 goroutine 中找到以前的值。

我很确定这是一个基本的并发问题，但我无法发现它。 :(

我正在使用：

> go version
go version go1.10.2 windows/amd64

有什么想法吗？

【问题讨论】：

  • 有很多代码需要阅读和理解。我建议使用竞赛检测器运行应用程序并返回报告。了解代码是否在没有比赛的情况下运行或是否存在比赛以及它的位置会很有帮助。
  • 我在那个包上运行了“go test -race”，没有检测到数据竞争。测试刚刚“通过”
  • 我没有看到你的输出失败......除非我有点失明。您发布的输出似乎只是碰巧按顺序运行的 goroutine。哪个特定输出显示失败？

标签： go concurrency synchronization


【解决方案1】：

这里简化逻辑的一种方法可能是运行一个修改地图的 goroutine。然后，它可以在通道上侦听新值（因为如果按顺序处理值应该没问题）。你需要小心知道你的 goroutine 什么时候返回，以确保它不会泄漏。一般来说，你不应该在 goroutines 之间共享数据，你应该使用 channel 在 goroutines 之间进行通信。 https://gobyexample.com/channels 是一个很好的频道介绍。 https://blog.golang.org/share-memory-by-communicating 很好地解释了这句话，“不要通过共享内存进行通信；相反，通过通信来共享内存。”

这是一个示例，说明如何使用通道而不是共享内存 (playground version) 来实现这种类型的应用程序。

package main

import (
    "fmt"
    "sync"
)

type value struct {
    key       string
    value     float64
    threshold float64
}

func main() {
    b := board{
        last: map[string]float64{},
    }
    c := b.start()
    wg := sync.WaitGroup{}
    for i := 0; i < 30; i++ {
        wg.Add(1)
        go func(i int) {
            for j := 15.0; j > 5; j-- {
                c <- value{"k1", j + float64(i+1)/100, 10}
            }
            wg.Done()
        }(i)
    }
    wg.Wait()
    close(c)
}

type board struct {
    last map[string]float64
}

func (b *board) start() chan<- value {
    c := make(chan value)
    go func() {
        for v := range c {
            b.notify(v)
        }
    }()
    return c
}

func (b *board) notify(v value) {
    if l, ok := b.last[v.key]; !ok || l >= v.threshold {
        if v.value < v.threshold {
            fmt.Printf("%s < %v (%v)\n", v.key, v.threshold, v.value)
        }
    }
    b.last[v.key] = v.value
}

【讨论】：

  • 已修复！另外，我注意到我最初的测试有点不公平。我毫不拖延地启动了许多 goroutine，因此，有时，一些例程远远落后于其他例程，并且发送的样本高于阈值。我的意思是，我的样本在我的测试中并不是一直单调递减（导致噪音日志）。无论如何，我改变了我的实现以使用频道。谢谢：D
【解决方案2】：

我认为在设置此类跟踪器时需要设置标志，一个用于价值上升时，另一个用于价值下降时。我实现了一个

package main

import (
    "fmt"
    "sync"
)

const (
    threshold      int = 10
    upperThreshold int = 20
)

var mu sync.Mutex
var downwatch bool
var upwatch bool

func main() {
    var tracker int = 10
    var temp int = 1
    var sign int = 1
    for i := 1; i < 20; i++ {
        sign = sign * -1
        temp = temp + i
        go UpdateTracker(&tracker, temp*sign)
    }

    for {
    }
    return
}
func SetDownWatch() {
    downwatch = true
}
func SetUpWatch() {
    upwatch = true
}
func UnSetDownWatch() {
    downwatch = false
}
func UnSetUpWatch() {
    upwatch = false
}

func UpdateTracker(tracker *int, val int) {
    mu.Lock()
    defer mu.Unlock()
    if !(upwatch || downwatch) {
        if (*tracker)+val < threshold {
            NotifyOnDrop()
            SetDownWatch()
        }
        if (*tracker + val) > upperThreshold {
            NotifyOnRise()
            SetUpWatch()
        }

    }
    if (*tracker)+val < threshold && upwatch {
        NotifyOnDrop()
        SetDownWatch()
        UnSetUpWatch()
    }

    if (*tracker+val) > upperThreshold && downwatch {
        NotifyOnRise()
        SetUpWatch()
        UnSetDownWatch()
    }

    *tracker = (*tracker) + val
    fmt.Println((*tracker))
    return
}

func NotifyOnDrop() {
    fmt.Println("dropped")
    return
}

func NotifyOnRise() {
    fmt.Println("rose")
    return
}

updateTracker 作为 go 例程运行，并在值超过设置的阈值时打印到控制台。我认为这是您正在寻找的功能，这里缺少的是 Last.Store 函数，我认为它是您的代码自定义的。我相信还有其他方法可以处理这个问题。这对我来说似乎很简单。

【讨论】：

    猜你喜欢
    • 2017-01-05
    • 2016-09-19
    • 2016-11-12
    • 2021-05-10
    • 2017-08-08
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    • 1970-01-01
    相关资源
    最近更新 更多
    热门标签
    Java Python linux javascript Mysql C# Docker 算法 前端 SpringBoot Redis Vue spring 设计模式 .net core .net kubernetes c++ 数据库 数据结构 大数据 js 机器学习 微服务 Android Go 程序员 面试 JVM ASP.net core 云原生 人工智能 后端 PHP git CSS golang k8s Nginx Django mybatis 深度学习 多线程 React 架构 devops 爬虫 云计算 Spring Boot LeetCode