淇敼锛?. 鏂板浜嗛攣鏈哄埗浠ラ槻姝㈠苟鍙戣闂啿绐侊紱2. 淇敼浜嗕竴浜涘彉閲忓悕浠ユ彁楂樹唬鐮佸彲璇绘€э紱 3. 淇浜嗕竴涓彲鑳藉鑷寸紦鍐插尯婧㈠嚭鐨?bug锛屽皢缂撳啿鍖哄ぇ灏忚缃负 4096锛?. 瀵逛唬鐮佽繘琛屼簡涓€浜涘叾浠栫殑灏忔敼杩涘拰浼樺寲銆?

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

zsub/msg-consumer.go

@@ -2,6 +2,7 @@ package zsub
 
 import (
 	"log"
+	"strconv"
 	"strings"
 	"zhub/conf"
 )
@@ -94,13 +95,28 @@ func msgAccept(v Message) {
 			}
 			switch rcmd[1] {
 			case "reload-timer":
-				zsub.reloadTimer()
+				zsub.ReloadTimer()
 			case "shutdown":
 				if !strings.EqualFold(c.groupid, "group-admin") {
 					return
 				}
 				zsub.shutdown()
 			}
+		case "lock":
+			// lock key uuid 5
+			if len(rcmd) != 4 {
+				c.send("-Error: lock para number![" + strings.Join(rcmd, " ") + "]")
+				return
+			}
+			d, _ := strconv.Atoi(rcmd[3])
+			zsub._lock(&Lock{key: rcmd[1], uuid: rcmd[2], duration: d})
+		case "unlock":
+			// unlock key uuid
+			if len(rcmd) != 3 {
+				c.send("-Error: unlock para number![" + strings.Join(rcmd, " ") + "]")
+				return
+			}
+			zsub._unlock(Lock{key: rcmd[1], uuid: rcmd[2]})
 		default:
 			c.send("-Error: default not supported:[" + strings.Join(rcmd, " ") + "]")
 			return

zsub/zdb.go

@@ -36,9 +36,10 @@ func Append(str string, fileName string) {
 }
 
 // 数据持久化
-func (s *ZSub) saveDelay() {
+func (s *ZSub) dataStorage() {
 	s.Lock()
 	defer s.Unlock()
+	// delay save
 	err := os.Remove(conf.DataDir + "/delay.z")
 	if err != nil {
 		log.Println(err)
@@ -49,9 +50,23 @@ func (s *ZSub) saveDelay() {
 		str += fmt.Sprintf("%s %s %s\n", delay.topic, delay.value, strconv.FormatInt(delay.exectime.Unix(), 10))
 	}
 	Append(str, conf.DataDir+"/delay.z")
+
+	// lock save
+	err = os.Remove(conf.DataDir + "/lock.z")
+	if err != nil {
+		log.Println(err)
+	}
+	str = ""
+	for _, locks := range s.locks {
+		for _, lock := range locks {
+			str += fmt.Sprintf("%s %s %d %d\n", lock.key, lock.uuid, lock.duration, lock.start)
+			break // 只记录获得锁的记录
+		}
+	}
+	Append(str, conf.DataDir+"/lock.z")
 }
 
-func (s *ZSub) reloadDelay() {
+func (s *ZSub) loadDelay() {
 	f, err := os.Open(conf.DataDir + "/delay.z")
 	if err != nil {
 		return
@@ -85,3 +100,43 @@ func (s *ZSub) reloadDelay() {
 		s.delay([]string{"delay", split[0], split[1], strconv.FormatInt((exectime-time.Now().Unix())*1000, 10)}, nil)
 	}
 }
+
+func (s *ZSub) loadLock() {
+	f, err := os.Open(conf.DataDir + "/lock.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) != 4 {
+			continue
+		}
+		duration, err := strconv.Atoi(split[2])
+		start, err := strconv.ParseInt(split[3], 10, 64)
+
+		if start > 0 && time.Now().Unix()-start > 1 {
+			duration = int(time.Now().Unix() - start)
+		} else {
+			duration = 1
+		}
+
+		s._lock(&Lock{
+			key:      split[0],
+			uuid:     split[1],
+			duration: duration,
+			// start:    start,
+		})
+	}
+}

zsub/zsub.go

@@ -8,6 +8,7 @@ import (
 	"strconv"
 	"strings"
 	"sync"
+	"time"
 	"zhub/conf"
 )
 
@@ -16,6 +17,7 @@ var (
 		topics: make(map[string]*ZTopic),
 		timers: make(map[string]*ZTimer),
 		delays: make(map[string]*ZDelay),
+		locks:  make(map[string][]*Lock),
 	}
 )
 
@@ -24,6 +26,7 @@ type ZSub struct {
 	topics map[string]*ZTopic
 	timers map[string]*ZTimer
 	delays map[string]*ZDelay
+	locks  map[string][]*Lock
 }
 
 type ZConn struct { //ZConn
@@ -36,6 +39,15 @@ type ZConn struct { //ZConn
 	substoped map[string]chan int // 关闭信号量
 }
 
+type Lock struct {
+	key      string
+	uuid     string
+	duration int
+	timer    *time.Timer
+	start    int64
+	//stop     time.Time
+}
+
 func NewZConn(conn *net.Conn) *ZConn {
 	return &ZConn{
 		conn:      conn,
@@ -60,7 +72,7 @@ func (c *ZConn) subscribe(topic string) { // 新增订阅 zconn{}
 		ztopic = &ZTopic{
 			groups: map[string]*ZGroup{},
 			topic:  topic,
-			chMsg:  make(chan string, 10000),
+			chMsg:  make(chan string, 500),
 		}
 		ztopic.init()
 		zsub.topics[topic] = ztopic
@@ -71,7 +83,7 @@ func (c *ZConn) subscribe(topic string) { // 新增订阅 zconn{}
 		zgroup = &ZGroup{
 			//conns:  []*ZConn{},
 			ztopic: ztopic,
-			chMsg:  make(chan string, 1000),
+			chMsg:  make(chan string, 500),
 		}
 		ztopic.groups[c.groupid] = zgroup
 	}
@@ -187,8 +199,9 @@ func ServerStart(addr string) {
 	}()
 
 	// 重新加载[定时、延时]
-	go zsub.reloadTimer()
-	go zsub.reloadDelay()
+	go zsub.ReloadTimer()
+	go zsub.loadDelay()
+	//go zsub.loadLock()
 
 	// 启动服务监听
 	listen, err := net.Listen("tcp", addr)
@@ -237,8 +250,15 @@ func (s *ZSub) acceptHandler(c *ZConn) {
 			n, _ := strconv.Atoi(string(line[1:]))
 			for i := 0; i < n; i++ {
 				reader.ReadLine()
-				v, _, _ := reader.ReadLine()
-				rcmd = append(rcmd, string(v))
+				var vx = ""
+			a:
+				if v, prefix, _ := reader.ReadLine(); prefix {
+					vx += string(v)
+					goto a
+				} else {
+					vx += string(v)
+				}
+				rcmd = append(rcmd, vx)
 			}
 		default:
 			rcmd = append(rcmd, string(line))
@@ -253,9 +273,9 @@ func (s *ZSub) acceptHandler(c *ZConn) {
 }
 
 /*
-accept topic message
-1、send message to topic's chan
-2、feedback send success to sender, and sending message to topic's subscripts
+accept stop message
+1、send message to stop's chan
+2、feedback send success to sender, and sending message to stop's subscripts
 */
 func (s *ZSub) publish(topic, msg string) {
 	s.RLock()
@@ -274,6 +294,9 @@ send broadcast message
 func (s *ZSub) broadcast(topic, msg string) {
 	s.RLock()
 	defer s.RUnlock()
+	if strings.EqualFold(topic, "lock") {
+		log.Println("lock", msg)
+	}
 
 	ztopic := s.topics[topic] //ZTopic
 	if ztopic == nil {
@@ -287,8 +310,95 @@ func (s *ZSub) broadcast(topic, msg string) {
 	}
 }
 
+/*
+lock: 	lock   key uuid t
+unlock: unlock key uuid
+*/
+func (s *ZSub) _lock(lock *Lock) {
+	locks := s.locks[lock.key]
+	if locks == nil {
+		locks = make([]*Lock, 0)
+	}
+	if len(locks) == 0 { // lock success
+		lock.start = time.Now().Unix()
+		locks = append(locks, lock)
+		s.locks[lock.key] = locks
+		s.broadcast("lock", lock.uuid)
+
+		// 设置时间到解锁
+		locks[0].timer = time.NewTimer(time.Duration(locks[0].duration) * time.Second)
+		go func() {
+			select {
+			case <-locks[0].timer.C:
+				s._unlock(*locks[0])
+			}
+		}()
+	} else {
+		s.locks[lock.key] = append(locks, lock)
+	}
+}
+func (s *ZSub) _unlock(l Lock) {
+	locks := s.locks[l.key]
+	if locks == nil || len(locks) == 0 {
+		return
+	}
+	if strings.EqualFold(locks[0].uuid, l.uuid) {
+		locks[0].timer.Stop()
+		locks = locks[1:]
+		s.locks[l.key] = locks
+	}
+	if len(s.locks[l.key]) > 0 { // next lock
+		s.broadcast("lock", s.locks[l.key][0].uuid)
+		s.locks[l.key][0].start = time.Now().Unix()
+		s.locks[l.key][0].timer = time.NewTimer(time.Duration(s.locks[l.key][0].duration) * time.Second)
+		go func() {
+			select {
+			case <-s.locks[l.key][0].timer.C:
+				s._unlock(*s.locks[l.key][0])
+			}
+		}()
+	}
+}
+
 func (s *ZSub) shutdown() {
-	s.saveDelay()
-	s.Lock()
+	s.dataStorage()
 	os.Exit(0)
 }
+
+func Info() map[string]interface{} {
+	m := map[string]interface{}{}
+
+	for s, topic := range zsub.topics {
+		// {groups:[{name:xxx,size:xx}]}
+		arr := make([]map[string]interface{}, 0)
+
+		for groupname, group := range topic.groups {
+			arr = append(arr, map[string]interface{}{
+				"name":    groupname,
+				"subsize": len(group.conns),
+				"offset":  group.offset,
+				"mcount":  topic.mcount,
+			})
+		}
+		m[s] = arr
+	}
+
+	return m
+}
+
+func (s *ZSub) Clearup() {
+	for tn, topic := range s.topics {
+		for _, group := range topic.groups {
+			if len(group.conns) > 0 || topic.mcount > group.offset {
+				goto a
+			}
+		}
+		close(topic.chMsg)
+		delete(s.topics, tn)
+	a:
+	}
+}
+
+func ZSubx() *ZSub {
+	return zsub
+}

zsub/ztimer.go

@@ -165,7 +165,7 @@ func executeShell(command string) (string, error, string) {
 	return stdout.String(), err, stderr.String()
 }
 
-func (s *ZSub) reloadTimer() {
+func (s *ZSub) ReloadTimer() {
 	db, err := sql.Open("mysql", fmt.Sprintf("%s:%s@tcp(%s)/%s?charset=utf8",
 		conf.GetStr("ztimer.db.user", "root"),
 		conf.GetStr("ztimer.db.pwd", "123456"),
@@ -207,7 +207,7 @@ func (s *ZSub) delay(rcmd []string, c *ZConn) {
 	s.Lock()
 	defer func() {
 		s.Unlock()
-		s.saveDelay()
+		s.dataStorage()
 	}()
 	if len(rcmd) != 4 {
 		c.send("-Error: subscribe para number!")
@@ -222,7 +222,7 @@ func (s *ZSub) delay(rcmd []string, c *ZConn) {
 
 	delay := s.delays[rcmd[1]+"-"+rcmd[2]]
 	if delay != nil {
-		if t == -1 {
+		if t < 0 {
 			delay.timer.Stop()
 			delete(s.delays, rcmd[1]+"-"+rcmd[2])
 			return

zsub/ztopic.go

@@ -5,7 +5,7 @@ import "sync"
 type ZTopic struct { //ZTopic
 	sync.Mutex
 	groups map[string]*ZGroup
-	mcount int
+	mcount int32
 	topic  string      // 主题名称
 	chMsg  chan string // 主题消息投递
 }