【发布时间】：2017-01-21 22:46:19
【问题描述】：
我有一些代码是作业调度程序，正在整理来自大量 TCP 套接字的大量数据。这段代码是Large number of transient objects - avoiding contention 方法的结果，它在很大程度上降低了 CPU 使用率，现在锁定也不是问题。
我的应用程序有时会锁定，并且“通道长度”日志是唯一不断重复的内容，因为数据仍在从我的套接字传入。但是，计数仍为 5000，并且没有进行下游处理。
我认为这个问题可能是一个竞争条件，它可能挂断的线路是select 中的channel <- msg jobDispatcher。问题是我不知道如何验证这一点。
我怀疑由于 select 可以随机获取项目，所以 goroutine 正在返回，而 shutdownChan 没有机会处理。然后数据命中 inboundFromTCP 并阻塞！
有人可能会在这里发现一些非常明显的错误。并希望提供解决方案！？
var MessageQueue = make(chan *trackingPacket_v1, 5000)
func init() {
go jobDispatcher(MessageQueue)
}
func addMessage(trackingPacket *trackingPacket_v1) {
// Send the packet to the buffered queue!
log.Println("Channel length:", len(MessageQueue))
MessageQueue <- trackingPacket
}
func jobDispatcher(inboundFromTCP chan *trackingPacket_v1) {
var channelMap = make(map[string]chan *trackingPacket_v1)
// Channel that listens for the strings that want to exit
shutdownChan := make(chan string)
for {
select {
case msg := <-inboundFromTCP:
log.Println("Got packet", msg.Avr)
channel, ok := channelMap[msg.Avr]
if !ok {
packetChan := make(chan *trackingPacket_v1)
channelMap[msg.Avr] = packetChan
go processPackets(packetChan, shutdownChan, msg.Avr)
packetChan <- msg
continue
}
channel <- msg
case shutdownString := <-shutdownChan:
log.Println("Shutting down:", shutdownString)
channel, ok := channelMap[shutdownString]
if ok {
delete(channelMap, shutdownString)
close(channel)
}
}
}
}
func processPackets(ch chan *trackingPacket_v1, shutdown chan string, id string) {
var messages = []*trackingPacket_v1{}
tickChan := time.NewTicker(time.Second * 1)
defer tickChan.Stop()
hasCheckedData := false
for {
select {
case msg := <-ch:
log.Println("Got a messages for", id)
messages = append(messages, msg)
hasCheckedData = false
case <-tickChan.C:
messages = cullChanMessages(messages)
if len(messages) == 0 {
messages = nil
shutdown <- id
return
}
// No point running checking when packets have not changed!!
if hasCheckedData == false {
processMLATCandidatesFromChan(messages)
hasCheckedData = true
}
case <-time.After(time.Duration(time.Second * 60)):
log.Println("This channel has been around for 60 seconds which is too much, kill it")
messages = nil
shutdown <- id
return
}
}
}
2016 年 1 月 20 日更新
我尝试将 channelMap 重新设计为具有一些互斥锁的全局变量，但最终还是死锁了。
稍微调整了代码，仍然锁定，但我不明白这个是怎么做的！！ https://play.golang.org/p/PGpISU4XBJ
2017 年 1 月 21 日更新 在一些建议之后，我将其放入一个独立的工作示例中，以便人们可以看到。 https://play.golang.org/p/88zT7hBLeD
这是一个长时间运行的过程，因此需要在机器上本地运行，因为游乐场会杀死它。希望这将有助于深入了解它！
【问题讨论】：
-
在死锁发生时获取堆栈跟踪，并查看每个 goroutine 被阻塞的位置。
-
我试过了，但是有 300 个左右的 goroutine，无法确定块在哪里！
-
是否有任何 pprof 命令对这里有帮助？
-
如果您在
inboundFromTCP的情况下，并且相应的processPackets关闭，它不会从地图中删除，因此您会遇到死锁。这有点过度使用频道的味道，但我没有清楚地了解问题，所以我不确定该向您发送哪个方向。 -
不要听那些讨厌频道的人。是的，通道就是锁，就像互斥锁一样。但是通道架构可以非常好。一旦你发现了你所看到的错误。