新增：延时事件持久化

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

app.go

@@ -4,6 +4,7 @@ import (
 	"log"
 	"os"
 	"strings"
+	"time"
 	"zhub/cli"
 	"zhub/conf"
 	"zhub/zsub"
@@ -29,11 +30,17 @@ func main() {
 	}
 
 	if len(os.Args) == 3 && strings.EqualFold(os.Args[1], "-r") {
-		if cli, err := cli.Create(addr, ""); err != nil {
+		if cli, err := cli.Create(addr, "group-admin"); err != nil {
 			log.Println(err)
 		} else {
-			cli.Cmd("reload-timer-config")
+			switch os.Args[2] {
+			case "timer":
+				cli.Cmd("reload-timer-config")
+			case "shutdown":
+				cli.Cmd("shutdown")
+			}
 			cli.Close()
+			time.Sleep(time.Millisecond * 10)
 		}
 		return
 	}

zsub/msg-consumer.go

@@ -8,10 +8,17 @@ import (
 	"zhub/conf"
 )
 
+type ZDelay struct {
+	topic    string
+	value    string
+	exectime time.Time
+	timer    *time.Timer
+}
+
 func msgAccept(v Message) {
 	defer func() {
 		if r := recover(); r != nil {
-			log.Println("ExecCmd Recovered:", r)
+			log.Println("msgAccept Recovered:", r)
 		}
 	}()
 	c := v.Conn
@@ -80,6 +87,11 @@ func msgAccept(v Message) {
 		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, " ") + "]")
@@ -90,7 +102,10 @@ func msgAccept(v Message) {
 // delay topic value 100 -> publish topic value
 func (s *ZSub) delay(rcmd []string, c *ZConn) {
 	s.Lock()
-	defer s.Unlock()
+	defer func() {
+		s.Unlock()
+		s.saveDelay()
+	}()
 	if len(rcmd) != 4 {
 		c.send("-Error: subscribe para number!")
 		return
@@ -102,20 +117,25 @@ func (s *ZSub) delay(rcmd []string, c *ZConn) {
 		return
 	}
 
-	timer := s.delays[rcmd[1]+"-"+rcmd[2]]
-	if timer != nil {
+	delay := s.delays[rcmd[1]+"-"+rcmd[2]]
+	if delay != nil {
 		if t == -1 {
-			timer.Stop()
+			delay.timer.Stop()
 			delete(s.delays, rcmd[1]+"-"+rcmd[2])
 			return
 		}
-		timer.Reset(time.Duration(t) * time.Millisecond)
+		delay.timer.Reset(time.Duration(t) * time.Millisecond)
 	} else {
-		timer = time.NewTimer(time.Duration(t) * time.Millisecond)
-		s.delays[rcmd[1]+"-"+rcmd[2]] = timer
+		delay := &ZDelay{
+			topic:    rcmd[1],
+			value:    rcmd[2],
+			exectime: time.Now().Add(time.Duration(t) * time.Millisecond),
+			timer:    time.NewTimer(time.Duration(t) * time.Millisecond),
+		}
+		s.delays[rcmd[1]+"-"+rcmd[2]] = delay
 		go func() {
 			select {
-			case <-timer.C:
+			case <-delay.timer.C:
 				zsub.publish(rcmd[1], rcmd[2])
 				delete(s.delays, rcmd[1]+"-"+rcmd[2])
 			}

zsub/zdb.go

@@ -1,7 +1,17 @@
 package zsub
 
+import (
+	"bufio"
+	"fmt"
+	"log"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+)
+
 var (
-	chanMessages = make(chan Message, 1000) //接收到的 所有消息数据
+// hubChan = make(chan Message, 1000) //接收到的 所有消息数据
 )
 
 // 数据封装
@@ -9,3 +19,66 @@ type Message struct {
 	Conn *ZConn
 	Rcmd []string
 }
+
+// 文件追加内容
+func Append(str string, fileName string) {
+	file, err := os.OpenFile(fileName, os.O_CREATE|os.O_WRONLY, os.ModeAppend)
+	if err != nil {
+		fmt.Println(err)
+	}
+	defer file.Close()
+
+	_, err = file.WriteString(str)
+	if err != nil {
+		log.Println(err)
+	}
+}
+
+// 数据持久化
+func (s *ZSub) saveDelay() {
+	s.Lock()
+	defer s.Unlock()
+	err := os.Remove("delay.z")
+	if err != nil {
+		log.Println(err)
+	}
+
+	for _, delay := range s.delays {
+		Append(fmt.Sprintf("%s %s %s\n", delay.topic, delay.value, strconv.FormatInt(delay.exectime.Unix(), 10)), "delay.z")
+	}
+}
+
+func (s *ZSub) reloadDelay() {
+	f, err := os.Open("delay.z")
+	if err != nil {
+		return
+	}
+	defer f.Close()
+
+	r := bufio.NewReader(f)
+	for {
+		bytes, err := r.ReadBytes('\n')
+		if err != nil {
+			return
+		}
+		line := string(bytes)
+		if len(line) == 0 {
+			continue
+		}
+		line = strings.Trim(line, " \r\n")
+		split := strings.Split(line, " ")
+		if len(split) != 3 {
+			continue
+		}
+
+		exectime, err := strconv.ParseInt(split[2], 10, 64)
+		if err != nil {
+			log.Println(err)
+			continue
+		}
+		if exectime < time.Now().Unix() {
+			continue
+		}
+		s.delay([]string{"delay", split[0], split[1], strconv.FormatInt((exectime-time.Now().Unix())*1000, 10)}, nil)
+	}
+}

zsub/zgroup.go

@@ -21,9 +21,10 @@ func (g *ZGroup) appendTo(c *ZConn) {
 		return
 	}
 
+	// create new goroutine consumer message
 	c.substoped[topic] = make(chan int, 0)
 	c.appendTo(g.conns)
-	go func() { // create new goroutine consumer message
+	go func() {
 		for {
 			select {
 			case msg, ok := <-g.chMsg:

zsub/zsub.go

@@ -4,17 +4,18 @@ import (
 	"bufio"
 	"log"
 	"net"
+	"os"
 	"strconv"
 	"strings"
 	"sync"
-	"time"
+	"zhub/conf"
 )
 
 var (
 	zsub = ZSub{
 		topics: make(map[string]*ZTopic),
 		timers: make(map[string]*ZTimer),
-		delays: make(map[string]*time.Timer),
+		delays: make(map[string]*ZDelay),
 	}
 )
 
@@ -22,7 +23,7 @@ type ZSub struct {
 	sync.RWMutex
 	topics map[string]*ZTopic
 	timers map[string]*ZTimer
-	delays map[string]*time.Timer
+	delays map[string]*ZDelay
 }
 
 type ZConn struct { //ZConn
@@ -190,8 +191,11 @@ func (c *ZConn) appendTo(arr []*ZConn) []*ZConn {
 */
 func ServerStart(addr string) {
 
-	// 加载定时调度服务
-	zsub.reloadTimerConfig()
+	conf.GetStr("data.dir", "data")
+
+	// 重新加载[定时、延时]
+	go zsub.reloadTimerConfig()
+	go zsub.reloadDelay()
 
 	// 启动服务监听
 	listen, err := net.Listen("tcp", addr)
@@ -200,19 +204,6 @@ func ServerStart(addr string) {
 	}
 	log.Printf("zhub started listen on: %s \n", addr)
 
-	// 启动消息监听处理
-	go func() {
-		for {
-			v, ok := <-chanMessages
-			if !ok {
-				break
-			}
-
-			// 事件消费
-			msgAccept(v)
-		}
-	}()
-
 	for {
 		conn, err := listen.Accept()
 		if err != nil {
@@ -264,7 +255,12 @@ func (s *ZSub) acceptHandler(c *ZConn) {
 			continue
 		}
 
-		// 接收消息 zdb fixme： 细节暴露太多
-		chanMessages <- Message{Conn: c, Rcmd: rcmd}
+		msgAccept(Message{Conn: c, Rcmd: rcmd})
 	}
 }
+
+func (s *ZSub) shutdown() {
+	s.saveDelay()
+	s.Lock()
+	os.Exit(0)
+}