修改：连接断开后消息推送携程未回收bug

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

cli_test.go

@@ -10,9 +10,9 @@ import (
 
 var (
 	//addr = "47.111.150.118:6066"
-	//addr = "127.0.0.1:1216"
+	addr = "127.0.0.1:1216"
 	//addr = "122.112.180.156:6066"
-	addr = "39.108.56.246:1216"
+	//addr = "39.108.56.246:1216"
 )
 
 func TestCli(t *testing.T) {
@@ -45,14 +45,14 @@ func TestTimer(t *testing.T) {
 	go func() {
 		client, _ := cli.Create(addr, "topic-1")
 
-		client.Subscribe("ax", func(v string) {
+		client.Subscribe("ax1", func(v string) {
 			log.Println("topic-1-ax: " + v)
 		})
 	}()
 	go func() {
 		client, _ := cli.Create(addr, "topic-1")
 
-		client.Subscribe("ax", func(v string) {
+		client.Subscribe("ax1", func(v string) {
 			log.Println("topic-2-ax: " + v)
 		})
 	}()
@@ -60,12 +60,12 @@ func TestTimer(t *testing.T) {
 	go func() {
 		client, _ := cli.Create(addr, "topic-1")
 
-		client.Subscribe("ax", func(v string) {
+		client.Subscribe("ax1", func(v string) {
 			log.Println("topic-3-ax: " + v)
 		})
 	}()
 
-	go func() {
+	/*go func() {
 		client, _ := cli.Create(addr, "topic-1")
 
 		client.Subscribe("ax", func(v string) {
@@ -78,7 +78,7 @@ func TestTimer(t *testing.T) {
 		client.Subscribe("ax", func(v string) {
 			log.Println("topic-5-ax: " + v)
 		})
-	}()
+	}()*/
 
 	/*go func() {
 		client, _ := cli.Create(addr, "topic-2")
@@ -121,9 +121,8 @@ func TestPublish(t *testing.T) {
 	if err != nil {
 		log.Println(err)
 	}
-	for i := 0; i < 100_0000; i++ {
-		client.Publish("ax", strconv.Itoa(i))
-		//time.Sleep(time.Second)
+	for i := 0; i < 10000; i++ {
+		client.Publish("ax1", strconv.Itoa(i))
 	}
 
 	time.Sleep(time.Second)

zsub/zgroup.go

@@ -17,17 +17,21 @@ func (g *ZGroup) appendTo(c *ZConn) {
 	c.appendTo(g.conns)
 	go func() { // 每个连接开启一个携程发送数据
 		for {
-			msg, ok := <-g.chMsg
-			if !ok {
-				break
-			}
+			select {
+			case msg, ok := <-g.chMsg:
+				if !ok {
+					return
+				}
 
-			err := c.send("message", g.ztopic.topic, msg)
-			if err != nil { // 失败处理
-				g.chMsg <- msg
-				break
+				err := c.send("message", g.ztopic.topic, msg)
+				if err != nil { // 失败处理
+					g.chMsg <- msg
+					return
+				}
+				atomic.AddInt32(&g.offset, 1)
+			case <-c.stoped:
+				return
 			}
-			atomic.AddInt32(&g.offset, 1)
 		}
 	}()
 }

zsub/zsub.go

@@ -28,6 +28,16 @@ type ZConn struct { //ZConn
 	groupid string
 	topics  []string
 	timers  []string // 订阅、定时调度分别创建各自连接
+	stoped  chan int // 关闭信号量
+}
+
+func NewZConn(conn *net.Conn) *ZConn {
+	return &ZConn{
+		conn:   conn,
+		topics: []string{},
+		timers: []string{},
+		stoped: make(chan int, 0),
+	}
 }
 
 /*
@@ -130,6 +140,7 @@ func (s *ZSub) broadcast(topic, msg string) {
 }
 
 func (s *ZSub) close(c *ZConn) {
+	close(c.stoped)
 	// sub
 	for _, topic := range c.topics {
 		s.unsubscribe(c, topic)
@@ -205,11 +216,8 @@ func ServerStart(addr string) {
 		}
 		log.Println("conn start: ", conn.RemoteAddr())
 
-		go zsub.acceptHandler(&ZConn{
-			conn:   &conn,
-			topics: []string{},
-			timers: []string{},
-		})
+		zConn := NewZConn(&conn)
+		go zsub.acceptHandler(zConn)
 	}
 }