修改:并发操作 map 崩溃

git-svn-id: svn://47.119.165.148/zhub@108 e63fbceb-bcc3-4977-ac22-735b83d8d0f4
This commit is contained in:
lxy
2021-02-22 11:14:03 +00:00
parent 6d79481ce0
commit d5ffd34b8b
2 changed files with 9 additions and 4 deletions

View File

@@ -14,6 +14,8 @@ type ZGroup struct { // ZGroup
}
func (g *ZGroup) appendTo(c *ZConn) {
c.Lock()
defer c.Unlock()
topic := g.ztopic.topic
// report subscribe topic check
@@ -22,7 +24,8 @@ func (g *ZGroup) appendTo(c *ZConn) {
}
// create new goroutine consumer message
c.substoped[topic] = make(chan int, 0)
unsubChan := make(chan int, 0)
c.substoped[topic] = unsubChan
c.appendTo(g.conns)
go func() {
for {
@@ -40,7 +43,9 @@ func (g *ZGroup) appendTo(c *ZConn) {
atomic.AddInt32(&g.offset, 1)
case <-c.stoped:
return
case <-c.substoped[topic]:
case <-unsubChan:
c.Lock()
defer c.Unlock()
delete(c.substoped, topic)
return
}

View File

@@ -90,8 +90,8 @@ func (s *ZSub) subscribe(c *ZConn, topic string) { // 新增订阅 zconn{}
取消订阅:
*/
func (s *ZSub) unsubscribe(c *ZConn, topic string) { // 取消订阅 zconn{}
s.Lock()
defer s.Unlock()
c.Lock()
defer c.Unlock()
close(c.substoped[topic])
ztopic := s.topics[topic] //ZTopic
if ztopic == nil {