修改：内部指令执行串行执行，避免并发下操作 map 崩溃

git-svn-id: svn://47.119.165.148/zhub@109 e63fbceb-bcc3-4977-ac22-735b83d8d0f4

zsub/msg-consumer.go

@@ -8,6 +8,8 @@ import (
 	"zhub/conf"
 )
 
+var funChan = make(chan func(), 1000)
+
 type ZDelay struct {
 	topic    string
 	value    string
@@ -56,46 +58,53 @@ func msgAccept(v Message) {
 	}
 
 	cmd := rcmd[0]
-	switch cmd {
+	if strings.EqualFold(cmd, "groupid") {
-	case "groupid":
 		c.groupid = rcmd[1]
-	case "subscribe":
+		return
-		// subscribe x y z
+	} else if strings.EqualFold(cmd, "publish") {
-		for _, topic := range rcmd[1:] {
-			zsub.subscribe(c, topic) // todo: 批量一次订阅
-		}
-	case "unsubscribe":
-		for _, topic := range rcmd[1:] {
-			zsub.unsubscribe(c, topic)
-		}
-	case "publish":
 		if len(rcmd) != 3 {
 			c.send("-Error: publish para number!")
 		} else {
 			zsub.publish(rcmd[1], rcmd[2])
 		}
-	case "broadcast":
+		return
-		zsub.broadcast(rcmd[1], rcmd[2])
+	}
-	case "delay":
+
-		zsub.delay(rcmd, c)
+	// 内部执行指令 加入执行队列
-	case "timer":
+	funChan <- func() {
-		zsub.timer(rcmd, c)
+		switch cmd {
-	case "cmd":
+		case "subscribe":
-		if len(rcmd) == 1 {
+			// subscribe x y z
-			return
+			for _, topic := range rcmd[1:] {
-		}
+				zsub.subscribe(c, topic) // todo: 批量一次订阅
-		switch rcmd[1] {
+			}
-		case "reload-timer-config":
+		case "unsubscribe":
-			zsub.reloadTimerConfig()
+			for _, topic := range rcmd[1:] {
-		case "shutdown":
+				zsub.unsubscribe(c, topic)
-			if !strings.EqualFold(c.groupid, "group-admin") {
+			}
+		case "broadcast":
+			zsub.broadcast(rcmd[1], rcmd[2])
+		case "delay":
+			zsub.delay(rcmd, c)
+		case "timer":
+			zsub.timer(rcmd, c)
+		case "cmd":
+			if len(rcmd) == 1 {
 				return
 			}
-			zsub.shutdown()
+			switch rcmd[1] {
+			case "reload-timer-config":
+				zsub.reloadTimerConfig()
+			case "shutdown":
+				if !strings.EqualFold(c.groupid, "group-admin") {
+					return
+				}
+				zsub.shutdown()
+			}
+		default:
+			c.send("-Error: default not supported:[" + strings.Join(rcmd, " ") + "]")
+			return
 		}
-	default:
-		c.send("-Error: default not supported:[" + strings.Join(rcmd, " ") + "]")
-		return
 	}
 }
 

zsub/zsub.go

@@ -190,9 +190,18 @@ func (c *ZConn) appendTo(arr []*ZConn) []*ZConn {
 2、启动服务监听
 */
 func ServerStart(addr string) {
-
 	conf.GetStr("data.dir", "data")
 
+	go func() {
+		for {
+			fun, ok := <-funChan
+			if !ok {
+				break
+			}
+			fun()
+		}
+	}()
+
 	// 重新加载[定时、延时]
 	go zsub.reloadTimerConfig()
 	go zsub.reloadDelay()