如何跨 goroutines 共享地图

Question

我正在尝试在 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... 必须发送另一个通知。当多个 goroutines 操纵我的结构时，我遇到了问题。我尝试使用 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}

Answer 1

在这里简化逻辑的一种方法可能是运行一个修改地图的 goroutine。然后，它可以监听通道上的新值（因为如果按顺序处理值应该没问题）。您需要小心知道您的 goroutine 何时返回以确保它不会泄漏。一般来说，你不应该在 goroutines 之间共享数据，你应该使用通道在 goroutines 之间进行通信。

这是一个示例，说明如何使用通道而不是共享内存（游乐场版本）来实现此类应用程序。

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

}

Answer 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我认为是您的代码自定义的功能。我相信还有其他方法可以处理这个问题。这对我来说似乎很简单。