fix bug

Author: LIOU2021

_example/main.go

@@ -26,29 +26,16 @@ func init() {
 func main() {
 	flag.Parse()
 
-	c1 := &redisBuff.Config{
-		SendBuff:       100,
-		MsgBatch:       5,
-		RunnerInterval: 5 * time.Second,
-		CacheName:      "message-test",
-		LockDuration:   3 * time.Second,
-		RLockDuration:  3 * time.Second,
-		ClearMsgFunc: func(msg []string) {
-			fmt.Println("from client: ", msg)
-		},
-		Debug: true,
-	}
-
+	c1 := newConfig("message-test-1")
 	s1 := redisBuff.New(c1)
 	closeRunner := s1.SendMsgRunner()
 
 	go testAdd(s1)
 	go loopAdd(s1)
 	go testReadList(s1)
 
-	c2 := *c1
-	c2.CacheName += "-2"
-	s2 := redisBuff.New(&c2)
+	c2 := newConfig("message-test-2")
+	s2 := redisBuff.New(c2)
 	closeRunner2 := s2.SendMsgRunner()
 
 	go testAdd(s2)
@@ -60,9 +47,24 @@ func main() {
 		fmt.Println("send closeRunner signal")
 		closeRunner <- true
 		closeRunner2 <- true
+		time.Sleep(50 * time.Millisecond)
 	}
 }
 
+func newConfig(name string) *redisBuff.Config {
+	return &redisBuff.Config{
+		SendBuff:       100,
+		MsgBatch:       5,
+		RunnerInterval: 5 * time.Second,
+		CacheName:      name,
+		LockDuration:   3 * time.Second,
+		RLockDuration:  3 * time.Second,
+		ClearMsgFunc: func(msg []string) {
+			fmt.Println("from client: ", msg)
+		},
+		// Debug: true,
+	}
+}
 func testAdd(b *redisBuff.Buff) {
 	var wg sync.WaitGroup
 	for i := 0; i < 30; i++ {
@@ -81,11 +83,6 @@ func testReadList(b *redisBuff.Buff) {
 			fmt.Println("from testReadList-1: ", b.List())
 		}()
 	}
-
-	for {
-		fmt.Println("from testReadList-2: ", b.List())
-		time.Sleep(500 * time.Millisecond)
-	}
 }
 
 func loopAdd(b *redisBuff.Buff) {

buff.go

@@ -102,6 +102,8 @@ type Buff struct {
 	debug          bool
 	readLockName   string
 	writeLockName  string
+	L              sync.RWMutex
+	close          bool
 }
 
 // execute runner
@@ -116,7 +118,11 @@ func (b *Buff) SendMsgRunner() chan<- bool {
 			case msg := <-b.send:
 				b.pushMsgWithLock(msg)
 			case <-done:
+				b.L.Lock()
 				close(b.send)
+				b.close = true
+				b.L.Unlock()
+
 				s := make(chan bool)
 				go func() { // 等chan上的讯息处理完后才能关闭
 					for len(b.send) > 0 {
@@ -162,7 +168,11 @@ func (b *Buff) getIntervalLock() (ok bool) {
 // add msg
 // sendBuff满时，将堵塞
 func (b *Buff) Add(data interface{}) {
-	b.send <- data
+	b.L.RLock()
+	defer b.L.RUnlock()
+	if !b.close {
+		b.send <- data
+	}
 }
 
 // 获取目前缓存上的资料
@@ -184,17 +194,17 @@ func (b *Buff) RLock() func() { // 怪怪的，打印频率有点怪
 		if rdb.Exists(ctx, b.writeLockName).Val() < 1 {
 			break
 		}
-		b.debugMsg("[%s RLock] - 等待写锁释放\n", b.cacheName)
+		// b.debugMsg("[%s RLock] - 等待写锁释放\n", b.cacheName)
 		time.Sleep(100 * time.Millisecond)
 	}
 
 	t := time.Now().Nanosecond()
 	rdb.Set(ctx, b.readLockName, t, b.rlockDuration)
-	b.debugMsg("[%s RLock] - 获得读锁\n", b.cacheName)
+	// b.debugMsg("[%s RLock] - 获得读锁\n", b.cacheName)
 	return func() {
 		v, _ := strconv.Atoi(rdb.Get(ctx, b.readLockName).Val())
 		if v == t {
-			b.debugMsg("[%s RLock] - 释放读锁\n", b.cacheName)
+			// b.debugMsg("[%s RLock] - 释放读锁\n", b.cacheName)
 			rdb.Del(ctx, b.readLockName)
 		}
 	}
@@ -206,21 +216,22 @@ func (b *Buff) lock() func() {
 		if rdb.Exists(ctx, b.readLockName).Val() < 1 {
 			break
 		}
-		b.debugMsg("[%s Lock] - 等待读锁释放\n", b.cacheName)
-		time.Sleep(100 * time.Second)
+		// b.debugMsg("[%s Lock] - 等待读锁释放\n", b.cacheName)
+		time.Sleep(100 * time.Millisecond)
 	}
 
 	for {
 		ok := rdb.SetNX(ctx, b.writeLockName, 1, b.lockDuration).Val()
 		if ok {
-			b.debugMsg("[%s Lock] - 获得写锁\n", b.cacheName)
+			// b.debugMsg("[%s Lock] - 获得写锁\n", b.cacheName)
 			break
 		}
+		// b.debugMsg("[%s Lock] - 读锁抢占失败\n", b.cacheName)
 		time.Sleep(time.Millisecond * 50)
 	}
 
 	return func() {
-		b.debugMsg("[%s Lock] - 释放写锁\n", b.cacheName)
+		// b.debugMsg("[%s Lock] - 释放写锁\n", b.cacheName)
 		rdb.Del(ctx, b.writeLockName)
 	}
 }