FIX:fatal error: concurrent map read and map write(#544)

.gitignore

@@ -3,4 +3,5 @@ go.mod
 go.sum
 vendor/
 coverage.txt
-examples/test
+examples/test
+/.vscode

consumer/offset_store.go

@@ -26,13 +26,12 @@ import (
 	"sync"
 	"time"
 
-	jsoniter "github.com/json-iterator/go"
-
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/internal/remote"
 	"github.com/apache/rocketmq-client-go/v2/internal/utils"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
 	"github.com/apache/rocketmq-client-go/v2/rlog"
+	jsoniter "github.com/json-iterator/go"
 )
 
 type readType int
@@ -101,7 +100,7 @@ func (mq *MessageQueueKey) UnmarshalText(text []byte) error {
 type localFileOffsetStore struct {
 	group       string
 	path        string
-	OffsetTable map[MessageQueueKey]int64
+	OffsetTable *sync.Map // concurrent safe , map[MessageQueueKey]int64
 	// mutex for offset file
 	mutex sync.Mutex
 }
@@ -110,7 +109,7 @@ func NewLocalFileOffsetStore(clientID, group string) OffsetStore {
 	store := &localFileOffsetStore{
 		group:       group,
 		path:        filepath.Join(_LocalOffsetStorePath, clientID, group, "offset.json"),
-		OffsetTable: make(map[MessageQueueKey]int64),
+		OffsetTable: new(sync.Map),
 	}
 	store.load()
 	return store
@@ -151,7 +150,9 @@ func (local *localFileOffsetStore) load() {
 	}
 
 	if datas != nil {
-		local.OffsetTable = datas
+		for k, v := range datas {
+			local.OffsetTable.Store(k, v)
+		}
 	}
 }
 
@@ -180,17 +181,17 @@ func (local *localFileOffsetStore) update(mq *primitive.MessageQueue, offset int
 		"new_offset":            offset,
 	})
 	key := MessageQueueKey(*mq)
-	localOffset, exist := local.OffsetTable[key]
+	localOffset, exist := local.OffsetTable.Load(key)
 	if !exist {
-		local.OffsetTable[key] = offset
+		local.OffsetTable.Store(key, offset)
 		return
 	}
 	if increaseOnly {
-		if localOffset < offset {
-			local.OffsetTable[key] = offset
+		if localOffset.(int64) < offset {
+			local.OffsetTable.Store(key, offset)
 		}
 	} else {
-		local.OffsetTable[key] = offset
+		local.OffsetTable.Store(key, offset)
 	}
 }
 
@@ -201,10 +202,17 @@ func (local *localFileOffsetStore) persist(mqs []*primitive.MessageQueue) {
 	local.mutex.Lock()
 	defer local.mutex.Unlock()
 
+	datas := make(map[MessageQueueKey]int64)
+	local.OffsetTable.Range(func(key, value interface{}) bool {
+		k := key.(MessageQueueKey)
+		v := value.(int64)
+		datas[k] = v
+		return true
+	})
+
 	wrapper := OffsetSerializeWrapper{
-		OffsetTable: local.OffsetTable,
+		OffsetTable: datas,
 	}
-
 	data, _ := jsoniter.Marshal(wrapper)
 	utils.CheckError(fmt.Sprintf("persist offset to %s", local.path), utils.WriteToFile(local.path, data))
 }
@@ -384,11 +392,11 @@ func (r *remoteBrokerOffsetStore) updateConsumeOffsetToBroker(group string, mq p
 	return r.client.InvokeOneWay(context.Background(), broker, cmd, 5*time.Second)
 }
 
-func readFromMemory(table map[MessageQueueKey]int64, mq *primitive.MessageQueue) int64 {
-	localOffset, exist := table[MessageQueueKey(*mq)]
+func readFromMemory(table *sync.Map, mq *primitive.MessageQueue) int64 {
+	localOffset, exist := table.Load(MessageQueueKey(*mq))
 	if !exist {
 		return -1
 	}
 
-	return localOffset
+	return localOffset.(int64)
 }

consumer/offset_store_test.go

@@ -21,12 +21,11 @@ import (
 	"path/filepath"
 	"testing"
 
-	"github.com/golang/mock/gomock"
-	. "github.com/smartystreets/goconvey/convey"
-
 	"github.com/apache/rocketmq-client-go/v2/internal"
 	"github.com/apache/rocketmq-client-go/v2/internal/remote"
 	"github.com/apache/rocketmq-client-go/v2/primitive"
+	"github.com/golang/mock/gomock"
+	. "github.com/smartystreets/goconvey/convey"
 )
 
 func TestNewLocalFileOffsetStore(t *testing.T) {
@@ -136,7 +135,7 @@ func TestLocalFileOffsetStore(t *testing.T) {
 			offset = localStore.read(mq, _ReadFromStore)
 			So(offset, ShouldEqual, 1)
 
-			delete(localStore.(*localFileOffsetStore).OffsetTable, MessageQueueKey(*mq))
+			localStore.(*localFileOffsetStore).OffsetTable.Delete(MessageQueueKey(*mq))
 			offset = localStore.read(mq, _ReadMemoryThenStore)
 			So(offset, ShouldEqual, 1)
 		})

go.sum

@@ -3,6 +3,7 @@ github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/emirpasic/gods v1.12.0 h1:QAUIPSaCu4G+POclxeqb3F+WPpdKqFGlw36+yOzGlrg=
 github.com/emirpasic/gods v1.12.0/go.mod h1:YfzfFFoVP/catgzJb4IKIqXjX78Ha8FMSDh3ymbK86o=
+github.com/feiquan123/rocketmq-client-go v1.2.4 h1:1GKQZ5pr1m4P2wruQfKtIbr0w9ZScoDefCqEOB/4E3c=
 github.com/golang/mock v1.3.1 h1:qGJ6qTW+x6xX/my+8YUVl4WNpX9B7+/l2tRsHGZ7f2s=
 github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=