Repository: apache/rocketmq-client-go
Pull request #200 (2019-09-10 06:10:41)

[ISSUE #185] Fix errors of comments and spelling
## What is the purpose of the change

Fix errors of comments and spelling

## Brief changelog

Fix errors of comments and spelling

## Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, `it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR`.

- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq/issues) filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue. 
- [x] Format the pull request title like `[ISSUE #123] Fix UnknownException when host config not exist`. Each commit in the pull request should have a meaningful subject line and body.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when a cross-module dependency exists.
- [ ] If this contribution is large, please file an [Apache Individual Contributor License Agreement](http://www.apache.org/licenses/#clas).


codecov-io: # [Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/200?src=pr&el=h1) Report
> Merging [#200](https://codecov.io/gh/apache/rocketmq-client-go/pull/200?src=pr&el=desc) into [native](https://codecov.io/gh/apache/rocketmq-client-go/commit/303ae26b2348e899930222d3a4e90e20b2531a17?src=pr&el=desc) will **not change** coverage.
> The diff coverage is `0%`.

[![Impacted file tree graph](https://codecov.io/gh/apache/rocketmq-client-go/pull/200/graphs/tree.svg?width=650&token=CiV9ofPkG3&height=150&src=pr)](https://codecov.io/gh/apache/rocketmq-client-go/pull/200?src=pr&el=tree)

```diff
@@           Coverage Diff           @@
##           native     #200   +/-   ##
=======================================
  Coverage   26.37%   26.37%           
=======================================
  Files          46       46           
  Lines        4671     4671           
=======================================
  Hits         1232     1232           
  Misses       3266     3266           
  Partials      173      173
```


| [Impacted Files](https://codecov.io/gh/apache/rocketmq-client-go/pull/200?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [primitive/ctx.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/200/diff?src=pr&el=tree#diff-cHJpbWl0aXZlL2N0eC5nbw==) | `0% <0%> (ø)` | :arrow_up: |
| [consumer/push\_consumer.go](https://codecov.io/gh/apache/rocketmq-client-go/pull/200/diff?src=pr&el=tree#diff-Y29uc3VtZXIvcHVzaF9jb25zdW1lci5nbw==) | `14.91% <0%> (ø)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/200?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/rocketmq-client-go/pull/200?src=pr&el=footer). Last update [303ae26...e9bd30b](https://codecov.io/gh/apache/rocketmq-client-go/pull/200?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/consumer/consumer.go b/consumer/consumer.go
index 55f94b5..0044bf6 100644
--- a/consumer/consumer.go
+++ b/consumer/consumer.go
@@ -811,16 +811,13 @@ func (dc *defaultConsumer) processPullResult(mq *primitive.MessageQueue, result
 
 		// TODO: add filter message hook
 		for _, msg := range msgListFilterAgain {
-			traFlag, _ := strconv.ParseBool(msg.Properties[primitive.PropertyTransactionPrepared])
+			traFlag, _ := strconv.ParseBool(msg.GetProperty(primitive.PropertyTransactionPrepared))
 			if traFlag {
-				msg.TransactionId = msg.Properties[primitive.PropertyUniqueClientMessageIdKeyIndex]
+				msg.TransactionId = msg.GetProperty(primitive.PropertyUniqueClientMessageIdKeyIndex)
 			}
 
-			if msg.Properties == nil {
-				msg.Properties = make(map[string]string)
-			}
-			msg.Properties[primitive.PropertyMinOffset] = strconv.FormatInt(result.MinOffset, 10)
-			msg.Properties[primitive.PropertyMaxOffset] = strconv.FormatInt(result.MaxOffset, 10)
+			msg.WithProperty(primitive.PropertyMinOffset, strconv.FormatInt(result.MinOffset, 10))
+			msg.WithProperty(primitive.PropertyMaxOffset, strconv.FormatInt(result.MaxOffset, 10))
 		}
 
 		result.SetMessageExts(msgListFilterAgain)
diff --git a/consumer/process_queue.go b/consumer/process_queue.go
index cff8a9f..19fbafa 100644
--- a/consumer/process_queue.go
+++ b/consumer/process_queue.go
@@ -101,7 +101,7 @@ func (pq *processQueue) putMessage(messages ...*primitive.MessageExt) {
 	}
 
 	msg := messages[len(messages)-1]
-	maxOffset, err := strconv.ParseInt(msg.Properties[primitive.PropertyMaxOffset], 10, 64)
+	maxOffset, err := strconv.ParseInt(msg.GetProperty(primitive.PropertyMaxOffset), 10, 64)
 	if err != nil {
 		acc := maxOffset - msg.QueueOffset
 		if acc > 0 {
@@ -172,7 +172,7 @@ func (pq *processQueue) cleanExpiredMsg(consumer defaultConsumer) {
 		}
 		_, firstValue := pq.msgCache.Min()
 		msg := firstValue.(*primitive.MessageExt)
-		startTime := msg.Properties[primitive.PropertyConsumeStartTime]
+		startTime := msg.GetProperty(primitive.PropertyConsumeStartTime)
 		if startTime != "" {
 			st, err := strconv.ParseInt(startTime, 10, 64)
 			if err != nil {
diff --git a/consumer/push_consumer.go b/consumer/push_consumer.go
index cef744a..8b32103 100644
--- a/consumer/push_consumer.go
+++ b/consumer/push_consumer.go
@@ -495,7 +495,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 
 		switch result.Status {
 		case primitive.PullFound:
-			rlog.Infof("Topic: %s, QueueId: %d found messages.", request.mq.Topic, request.mq.QueueId)
+			rlog.Debugf("Topic: %s, QueueId: %d found messages.", request.mq.Topic, request.mq.QueueId)
 			prevRequestOffset := request.nextOffset
 			request.nextOffset = result.NextBeginOffset
 
@@ -516,7 +516,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 					"firstMsgOffset=%d, prevRequestOffset=%d]", result.NextBeginOffset, firstMsgOffset, prevRequestOffset)
 			}
 		case primitive.PullNoNewMsg:
-			rlog.Infof("Topic: %s, QueueId: %d no more msg, current offset: %d, next offset: %d", request.mq.Topic, request.mq.QueueId, pullRequest.QueueOffset, result.NextBeginOffset)
+			rlog.Debugf("Topic: %s, QueueId: %d no more msg, current offset: %d, next offset: %d", request.mq.Topic, request.mq.QueueId, pullRequest.QueueOffset, result.NextBeginOffset)
 		case primitive.PullNoMsgMatched:
 			request.nextOffset = result.NextBeginOffset
 			pc.correctTagsOffset(request)
@@ -677,14 +677,12 @@ func (pc *pushConsumer) resetRetryAndNamespace(subMsgs []*primitive.MessageExt)
 	beginTime := time.Now()
 	for idx := range subMsgs {
 		msg := subMsgs[idx]
-		if msg.Properties != nil {
-			retryTopic := msg.Properties[primitive.PropertyRetryTopic]
-			if retryTopic == "" && groupTopic == msg.Topic {
-				msg.Topic = retryTopic
-			}
-			subMsgs[idx].Properties[primitive.PropertyConsumeStartTime] = strconv.FormatInt(
-				beginTime.UnixNano()/int64(time.Millisecond), 10)
+		retryTopic := msg.GetProperty(primitive.PropertyRetryTopic)
+		if retryTopic == "" && groupTopic == msg.Topic {
+			msg.Topic = retryTopic
 		}
+		subMsgs[idx].WithProperty(primitive.PropertyConsumeStartTime, strconv.FormatInt(
+			beginTime.UnixNano()/int64(time.Millisecond), 10))
 	}
 }
 
@@ -734,7 +732,7 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.
 
 			consumeRT := time.Now().Sub(beginTime)
 			if err != nil {
-				msgCtx.Properties[primitive.PropCtxType] = string(primitive.ExceptionRetrun)
+				msgCtx.Properties[primitive.PropCtxType] = string(primitive.ExceptionReturn)
 			} else if consumeRT >= pc.option.ConsumeTimeout {
 				msgCtx.Properties[primitive.PropCtxType] = string(primitive.TimeoutReturn)
 			} else if result == ConsumeSuccess {
@@ -805,12 +803,12 @@ func (pc *pushConsumer) consumeMessageOrderly(pq *processQueue, mq *primitive.Me
 			if pc.model == Clustering {
 				if !pq.locked {
 					rlog.Warn("the message queue not locked, so consume later: %v", mq)
-					pc.tryLocakLaterAndReconsume(mq, 10)
+					pc.tryLockLaterAndReconsume(mq, 10)
 					return
 				}
 				if pq.isLockExpired() {
 					rlog.Warn("the message queue lock expired, so consume later: %v", mq)
-					pc.tryLocakLaterAndReconsume(mq, 10)
+					pc.tryLockLaterAndReconsume(mq, 10)
 					return
 				}
 			}
@@ -909,7 +907,7 @@ func (pc *pushConsumer) consumeMessageOrderly(pq *processQueue, mq *primitive.Me
 		if pq.dropped {
 			rlog.Warn("the message queue not be able to consume, because it's dropped. %v", mq)
 		}
-		pc.tryLocakLaterAndReconsume(mq, 100)
+		pc.tryLockLaterAndReconsume(mq, 100)
 	}
 }
 
@@ -920,7 +918,7 @@ func (pc *pushConsumer) checkReconsumeTimes(msgs []*primitive.MessageExt) bool {
 		for _, msg := range msgs {
 			if msg.ReconsumeTimes > maxReconsumeTimes {
 				rlog.Warn("msg will be send to retry topic due to ReconsumeTimes > %d, \n", maxReconsumeTimes)
-				msg.Properties["RECONSUME_TIME"] = strconv.Itoa(int(msg.ReconsumeTimes))
+				msg.WithProperty("RECONSUME_TIME", strconv.Itoa(int(msg.ReconsumeTimes)))
 				if !pc.sendMessageBack("", msg, -1) {
 					suspend = true
 					msg.ReconsumeTimes += 1
@@ -950,7 +948,7 @@ func (pc *pushConsumer) getMaxReconsumeTimes() int32 {
 	}
 }
 
-func (pc *pushConsumer) tryLocakLaterAndReconsume(mq *primitive.MessageQueue, delay int64) {
+func (pc *pushConsumer) tryLockLaterAndReconsume(mq *primitive.MessageQueue, delay int64) {
 	time.Sleep(time.Duration(delay) * time.Millisecond)
 	if pc.lock(mq) == true {
 		pc.submitConsumeRequestLater(10)
diff --git a/examples/consumer/retry/concurrent/main.go b/examples/consumer/retry/concurrent/main.go
index cb4b127..8ca31b0 100644
--- a/examples/consumer/retry/concurrent/main.go
+++ b/examples/consumer/retry/concurrent/main.go
@@ -15,13 +15,6 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-/**
- * use concurrent consumer model, when Subscribe function return consumer.ConsumeRetryLater, the message will be
- * send to rocketmq retry topic. we could set DelayLevelWhenNextConsume in ConsumeConcurrentlyContext, which used to
- * indicate the delay of message re-send to origin topic from retry topic.
- * in this example, we always set DelayLevelWhenNextConsume=1, means that the message will be sent to origin topic after
- * 1s. in case of the unlimit retry, we will return consumer.ConsumeSuccess after ReconsumeTimes > 5
- */
 package main
 
 import (
@@ -35,6 +28,12 @@ import (
 	"github.com/apache/rocketmq-client-go/primitive"
 )
 
+// use concurrent consumer model, when Subscribe function return consumer.ConsumeRetryLater, the message will be
+// send to RocketMQ retry topic. we could set DelayLevelWhenNextConsume in ConsumeConcurrentlyContext, which used to
+// indicate the delay of message re-send to origin topic from retry topic.
+//
+// in this example, we always set DelayLevelWhenNextConsume=1, means that the message will be sent to origin topic after
+// 1s. in case of the unlimited retry, we will return consumer.ConsumeSuccess after ReconsumeTimes > 5
 func main() {
 	c, _ := rocketmq.NewPushConsumer(
 		consumer.WithGroupName("testGroup"),
@@ -42,13 +41,18 @@ func main() {
 		consumer.WithConsumerModel(consumer.Clustering),
 	)
 
-	retryLevel := 1 // mean 10s later before consume
+	// The DelayLevel specify the waiting time that before next reconsume,
+	// and it range is from 1 to 18 now.
+	//
+	// The time of each level is the value of indexing of {level-1} in [1s, 5s, 10s, 30s,
+	// 1m, 2m, 3m, 4m, 5m, 6m, 7m, 8m, 9m, 10m, 20m, 30m, 1h, 2h]
+	delayLevel := 1
 	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
 		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
 		fmt.Printf("subscribe callback len: %d \n", len(msgs))
 
 		concurrentCtx, _ := primitive.GetConcurrentlyCtx(ctx)
-		concurrentCtx.DelayLevelWhenNextConsume = retryLevel // only run when return consumer.ConsumeRetryLater
+		concurrentCtx.DelayLevelWhenNextConsume = delayLevel // only run when return consumer.ConsumeRetryLater
 
 		for _, msg := range msgs {
 			if msg.ReconsumeTimes > 5 {
diff --git a/examples/consumer/simple/main.go b/examples/consumer/simple/main.go
index bd2058e..33ff6ee 100644
--- a/examples/consumer/simple/main.go
+++ b/examples/consumer/simple/main.go
@@ -33,7 +33,7 @@ func main() {
 		consumer.WithGroupName("testGroup"),
 		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
 	)
-	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
+	err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx context.Context,
 		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
 		fmt.Printf("subscribe callback: %v \n", msgs)
 		return consumer.ConsumeSuccess, nil
diff --git a/examples/producer/acl/main.go b/examples/producer/acl/main.go
index a98500c..2bd97fb 100644
--- a/examples/producer/acl/main.go
+++ b/examples/producer/acl/main.go
@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"strconv"
 
 	"github.com/apache/rocketmq-client-go"
 	"github.com/apache/rocketmq-client-go/primitive"
@@ -50,11 +49,8 @@ func main() {
 		os.Exit(1)
 	}
 	for i := 0; i < 100000; i++ {
-		res, err := p.SendSync(context.Background(), &primitive.Message{
-			Topic:      "test",
-			Body:       []byte("Hello RocketMQ Go Client!"),
-			Properties: map[string]string{"order": strconv.Itoa(i)},
-		})
+		res, err := p.SendSync(context.Background(), primitive.NewMessage("test",
+			[]byte("Hello RocketMQ Go Client!")))
 
 		if err != nil {
 			fmt.Printf("send message error: %s\n", err)
diff --git a/examples/producer/async/main.go b/examples/producer/async/main.go
index ff3934f..01f7f05 100644
--- a/examples/producer/async/main.go
+++ b/examples/producer/async/main.go
@@ -21,7 +21,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"strconv"
 	"sync"
 
 	"github.com/apache/rocketmq-client-go"
@@ -52,11 +51,7 @@ func main() {
 					fmt.Printf("send message success: result=%s\n", result.String())
 				}
 				wg.Done()
-			}, &primitive.Message{
-				Topic:      "TopicTest",
-				Body:       []byte("Hello RocketMQ Go Client!"),
-				Properties: map[string]string{"id": strconv.Itoa(i)},
-			})
+			}, primitive.NewMessage("test", []byte("Hello RocketMQ Go Client!")))
 
 		if err != nil {
 			fmt.Printf("send message error: %s\n", err)
diff --git a/examples/producer/delay/main.go b/examples/producer/delay/main.go
index 4466e8d..61ff29a 100644
--- a/examples/producer/delay/main.go
+++ b/examples/producer/delay/main.go
@@ -38,10 +38,7 @@ func main() {
 		os.Exit(1)
 	}
 	for i := 0; i < 10; i++ {
-		msg := &primitive.Message{
-			Topic: "TopicTest",
-			Body:  []byte("Hello RocketMQ Go Client!"),
-		}
+		msg := primitive.NewMessage("test", []byte("Hello RocketMQ Go Client!"))
 		msg.WithDelayTimeLevel(3)
 		res, err := p.SendSync(context.Background(), msg)
 
diff --git a/examples/producer/interceptor/main.go b/examples/producer/interceptor/main.go
index 643d2a8..22fc552 100644
--- a/examples/producer/interceptor/main.go
+++ b/examples/producer/interceptor/main.go
@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"strconv"
 
 	"github.com/apache/rocketmq-client-go"
 	"github.com/apache/rocketmq-client-go/primitive"
@@ -41,11 +40,8 @@ func main() {
 		os.Exit(1)
 	}
 	for i := 0; i < 10; i++ {
-		res, err := p.SendSync(context.Background(), &primitive.Message{
-			Topic:      "TopicTest",
-			Body:       []byte("Hello RocketMQ Go Client!"),
-			Properties: map[string]string{"order": strconv.Itoa(i)},
-		})
+		res, err := p.SendSync(context.Background(), primitive.NewMessage("test",
+			[]byte("Hello RocketMQ Go Client!")))
 
 		if err != nil {
 			fmt.Printf("send message error: %s\n", err)
diff --git a/examples/producer/namespace/main.go b/examples/producer/namespace/main.go
index 002868a..881c688 100644
--- a/examples/producer/namespace/main.go
+++ b/examples/producer/namespace/main.go
@@ -22,7 +22,6 @@ import (
 	"context"
 	"fmt"
 	"os"
-	"strconv"
 
 	"github.com/apache/rocketmq-client-go"
 	"github.com/apache/rocketmq-client-go/primitive"
@@ -51,11 +50,8 @@ func main() {
 		os.Exit(1)
 	}
 	for i := 0; i < 100000; i++ {
-		res, err := p.SendSync(context.Background(), &primitive.Message{
-			Topic:      "test",
-			Body:       []byte("Hello RocketMQ Go Client!"),
-			Properties: map[string]string{"order": strconv.Itoa(i)},
-		})
+		res, err := p.SendSync(context.Background(), primitive.NewMessage("test",
+			[]byte("Hello RocketMQ Go Client!")))
 
 		if err != nil {
 			fmt.Printf("send message error: %s\n", err)
diff --git a/examples/producer/simple/main.go b/examples/producer/simple/main.go
index fc9f77f..0a99835 100644
--- a/examples/producer/simple/main.go
+++ b/examples/producer/simple/main.go
@@ -39,10 +39,8 @@ func main() {
 		os.Exit(1)
 	}
 	for i := 0; i < 1000; i++ {
-		res, err := p.SendSync(context.Background(), &primitive.Message{
-			Topic: "TopicTest",
-			Body:  []byte("Hello RocketMQ Go Client!"),
-		})
+		res, err := p.SendSync(context.Background(), primitive.NewMessage("test",
+			[]byte("Hello RocketMQ Go Client!")))
 
 		if err != nil {
 			fmt.Printf("send message error: %s\n", err)
diff --git a/examples/producer/tag/main.go b/examples/producer/tag/main.go
index 6b22454..ca408ab 100644
--- a/examples/producer/tag/main.go
+++ b/examples/producer/tag/main.go
@@ -40,10 +40,8 @@ func main() {
 	tags := []string{"TagA", "TagB", "TagC"}
 	for i := 0; i < 3; i++ {
 		tag := tags[i%3]
-		msg := &primitive.Message{
-			Topic: "TopicTest",
-			Body:  []byte("Hello RocketMQ Go Client!"),
-		}
+		msg := primitive.NewMessage("test",
+			[]byte("Hello RocketMQ Go Client!"))
 		msg.WithTag(tag)
 
 		res, err := p.SendSync(context.Background(), msg)
diff --git a/examples/producer/trace/main.go b/examples/producer/trace/main.go
index e8fd386..c375c6a 100644
--- a/examples/producer/trace/main.go
+++ b/examples/producer/trace/main.go
@@ -45,10 +45,8 @@ func main() {
 		os.Exit(1)
 	}
 	for i := 0; i < 1; i++ {
-		res, err := p.SendSync(context.Background(), &primitive.Message{
-			Topic: "TopicTest",
-			Body:  []byte("Hello RocketMQ Go Client!"),
-		})
+		res, err := p.SendSync(context.Background(), primitive.NewMessage("test",
+			[]byte("Hello RocketMQ Go Client!")))
 
 		if err != nil {
 			fmt.Printf("send message error: %s\n", err)
diff --git a/internal/client.go b/internal/client.go
index cb4cfb6..99227ef 100644
--- a/internal/client.go
+++ b/internal/client.go
@@ -191,12 +191,12 @@ func GetOrNewRocketMQClient(option ClientOptions, callbackCh chan interface{}) *
 			}
 			msgExt := msgExts[0]
 			// TODO: add namespace support
-			transactionID := msgExt.Properties[primitive.PropertyUniqueClientMessageIdKeyIndex]
+			transactionID := msgExt.GetProperty(primitive.PropertyUniqueClientMessageIdKeyIndex)
 			if len(transactionID) > 0 {
 				msgExt.TransactionId = transactionID
 			}
-			group, existed := msgExt.Properties[primitive.PropertyProducerGroup]
-			if !existed {
+			group := msgExt.GetProperty(primitive.PropertyProducerGroup)
+			if group == "" {
 				rlog.Warn("checkTransactionState, pick producer group failed")
 				return nil
 			}
@@ -442,7 +442,7 @@ func (c *rmqClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingC
 
 	msgIDs := make([]string, 0)
 	for i := 0; i < len(msgs); i++ {
-		msgIDs = append(msgIDs, msgs[i].Properties[primitive.PropertyUniqueClientMessageIdKeyIndex])
+		msgIDs = append(msgIDs, msgs[i].GetProperty(primitive.PropertyUniqueClientMessageIdKeyIndex))
 	}
 
 	regionId := cmd.ExtFields[primitive.PropertyMsgRegion]
diff --git a/internal/trace.go b/internal/trace.go
index 3bceefb..2aa56f9 100644
--- a/internal/trace.go
+++ b/internal/trace.go
@@ -432,7 +432,7 @@ func (td *traceDispatcher) buildSendRequest(mq *primitive.MessageQueue,
 		QueueId:       mq.QueueId,
 		BornTimestamp: time.Now().UnixNano() / int64(time.Millisecond),
 		Flag:          msg.Flag,
-		Properties:    primitive.MarshalPropeties(msg.Properties),
+		Properties:    msg.MarshallProperties(),
 	}
 
 	return remote.NewRemotingCommand(ReqSendMessage, req, msg.Body)
diff --git a/primitive/ctx.go b/primitive/ctx.go
index 186521a..0937498 100644
--- a/primitive/ctx.go
+++ b/primitive/ctx.go
@@ -39,7 +39,7 @@ func (c ConsumeReturnType) Ordinal() int {
 		return 0
 	case TimeoutReturn:
 		return 1
-	case ExceptionRetrun:
+	case ExceptionReturn:
 		return 2
 	case NullReturn:
 		return 3
@@ -69,7 +69,7 @@ const (
 	PropCtxType                       = "ConsumeContextType"
 	SuccessReturn   ConsumeReturnType = "SUCCESS"
 	TimeoutReturn   ConsumeReturnType = "TIMEOUT"
-	ExceptionRetrun ConsumeReturnType = "EXCEPTION"
+	ExceptionReturn ConsumeReturnType = "EXCEPTION"
 	NullReturn      ConsumeReturnType = "RETURNNULL"
 	FailedReturn    ConsumeReturnType = "FAILED"
 )
diff --git a/primitive/message.go b/primitive/message.go
index 9e6ae02..2e81502 100644
--- a/primitive/message.go
+++ b/primitive/message.go
@@ -66,79 +66,119 @@ type Message struct {
 	Topic         string
 	Body          []byte
 	Flag          int32
-	Properties    map[string]string
 	TransactionId string
 	Batch         bool
-
 	// QueueID is the queue that messages will be sent to. the value must be set if want to custom the queue of message,
 	// just ignore if not.
 	Queue *MessageQueue
+
+	properties map[string]string
+	mutex      sync.RWMutex
 }
 
-func NewMessage(topic string, body []byte) *Message {
-	msg := &Message{
-		Topic:      topic,
-		Body:       body,
-		Properties: make(map[string]string),
+func (m *Message) WithProperty(key, value string) {
+	if key == "" || value == "" {
+		return
 	}
-	msg.Properties[PropertyWaitStoreMsgOk] = strconv.FormatBool(true)
-	return msg
+	m.mutex.Lock()
+	if m.properties == nil {
+		m.properties = make(map[string]string)
+	}
+	m.properties[key] = value
+	m.mutex.Unlock()
 }
 
-// WithDelayTimeLevel set message delay time to consume.
-// reference delay level definition: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
-// delay level starts from 1. for example, if we set param level=1, then the delay time is 1s.
-func (msg *Message) WithDelayTimeLevel(level int) *Message {
-	if msg.Properties == nil {
-		msg.Properties = make(map[string]string)
-	}
-	msg.Properties[PropertyDelayTimeLevel] = strconv.Itoa(level)
-	return msg
+func (m *Message) GetProperty(key string) string {
+	m.mutex.RLock()
+	v := m.properties[key]
+	m.mutex.RUnlock()
+	return v
 }
 
-func (msg *Message) WithTag(tags string) *Message {
-	if msg.Properties == nil {
-		msg.Properties = make(map[string]string)
+func (m *Message) RemoveProperty(key string) string {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	value, exist := m.properties[key]
+	if !exist {
+		return ""
 	}
-	msg.Properties[PropertyTags] = tags
-	return msg
+	delete(m.properties, key)
+	return value
+}
+func (m *Message) MarshallProperties() string {
+	m.mutex.RLock()
+	defer m.mutex.RUnlock()
+	buffer := bytes.NewBufferString("")
+	for k, v := range m.properties {
+		buffer.WriteString(k)
+		buffer.WriteRune(nameValueSeparator)
+		buffer.WriteString(v)
+		buffer.WriteRune(propertySeparator)
+	}
+	return buffer.String()
 }
 
-func (msg *Message) String() string {
-	return fmt.Sprintf("[topic=%s, body=%s, Flag=%d, Properties=%v, TransactionId=%s]",
-		msg.Topic, string(msg.Body), msg.Flag, msg.Properties, msg.TransactionId)
+// unmarshalProperties parse data into property kv pairs.
+func (m *Message) UnmarshalProperties(data []byte) {
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
+	if m.properties == nil {
+		m.properties = make(map[string]string)
+	}
+	items := bytes.Split(data, []byte{propertySeparator})
+	for _, item := range items {
+		kv := bytes.Split(item, []byte{nameValueSeparator})
+		if len(kv) == 2 {
+			m.properties[string(kv[0])] = string(kv[1])
+		}
+	}
 }
 
-func (msg *Message) WithProperty(key, value string) *Message {
-	msg.Properties[key] = value
+func NewMessage(topic string, body []byte) *Message {
+	msg := &Message{
+		Topic:      topic,
+		Body:       body,
+		properties: make(map[string]string),
+	}
+	//msg.properties[PropertyWaitStoreMsgOk] = strconv.FormatBool(true)
 	return msg
 }
 
-func (msg *Message) RemoveProperty(key string) string {
-	value, exist := msg.Properties[key]
-	if !exist {
-		return ""
-	}
+// WithDelayTimeLevel set message delay time to consume.
+// reference delay level definition: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
+// delay level starts from 1. for example, if we set param level=1, then the delay time is 1s.
+func (m *Message) WithDelayTimeLevel(level int) *Message {
+	m.WithProperty(PropertyDelayTimeLevel, strconv.Itoa(level))
+	return m
+}
 
-	delete(msg.Properties, key)
-	return value
+func (m *Message) WithTag(tags string) *Message {
+	m.WithProperty(PropertyTags, tags)
+	return m
 }
 
-func (msg *Message) WithKeys(keys []string) *Message {
+func (m *Message) WithKeys(keys []string) *Message {
 	var sb strings.Builder
 	for _, k := range keys {
 		sb.WriteString(k)
 		sb.WriteString(PropertyKeySeparator)
 	}
-	return msg.WithProperty(PropertyKeys, sb.String())
+
+	m.WithProperty(PropertyKeys, sb.String())
+	return m
+}
+
+func (m *Message) GetTags() string {
+	return m.GetProperty(PropertyTags)
 }
 
-func (msg *Message) GetTags() string {
-	return msg.Properties[PropertyTags]
+func (m *Message) GetKeys() string {
+	return m.GetProperty(PropertyKeys)
 }
 
-func (msg *Message) GetKeys() string {
-	return msg.Properties[PropertyKeys]
+func (m *Message) String() string {
+	return fmt.Sprintf("[topic=%s, body=%s, Flag=%d, properties=%v, TransactionId=%s]",
+		m.Topic, string(m.Body), m.Flag, m.properties, m.TransactionId)
 }
 
 type MessageExt struct {
@@ -159,15 +199,15 @@ type MessageExt struct {
 }
 
 func (msgExt *MessageExt) GetTags() string {
-	return msgExt.Properties[PropertyTags]
+	return msgExt.GetProperty(PropertyTags)
 }
 
 func (msgExt *MessageExt) GetRegionID() string {
-	return msgExt.Properties[PropertyMsgRegion]
+	return msgExt.GetProperty(PropertyMsgRegion)
 }
 
 func (msgExt *MessageExt) IsTraceOn() string {
-	return msgExt.Properties[PropertyTraceSwitch]
+	return msgExt.GetProperty(PropertyTraceSwitch)
 }
 
 func (msgExt *MessageExt) String() string {
@@ -265,14 +305,14 @@ func DecodeMessage(data []byte) []*MessageExt {
 		var propertiesLength int16
 		binary.Read(buf, binary.BigEndian, &propertiesLength)
 		if propertiesLength > 0 {
-			msg.Properties = unmarshalProperties(buf.Next(int(propertiesLength)))
+			msg.UnmarshalProperties(buf.Next(int(propertiesLength)))
 		}
 		count += 2 + int(propertiesLength)
 
 		msg.MsgId = createMessageId(hostBytes, port, msg.CommitLogOffset)
 		//count += 16
-		if msg.Properties == nil {
-			msg.Properties = make(map[string]string, 0)
+		if msg.properties == nil {
+			msg.properties = make(map[string]string, 0)
 		}
 		msgs = append(msgs, msg)
 	}
@@ -280,34 +320,6 @@ func DecodeMessage(data []byte) []*MessageExt {
 	return msgs
 }
 
-// unmarshalProperties parse data into property kv pairs.
-func unmarshalProperties(data []byte) map[string]string {
-	m := make(map[string]string)
-	items := bytes.Split(data, []byte{propertySeparator})
-	for _, item := range items {
-		kv := bytes.Split(item, []byte{nameValueSeparator})
-		if len(kv) == 2 {
-			m[string(kv[0])] = string(kv[1])
-		}
-	}
-	return m
-}
-
-func MarshalPropeties(properties map[string]string) string {
-	if properties == nil {
-		return ""
-	}
-	buffer := bytes.NewBufferString("")
-
-	for k, v := range properties {
-		buffer.WriteString(k)
-		buffer.WriteRune(nameValueSeparator)
-		buffer.WriteString(v)
-		buffer.WriteRune(propertySeparator)
-	}
-	return buffer.String()
-}
-
 // MessageQueue message queue
 type MessageQueue struct {
 	Topic      string `json:"topic"`
diff --git a/producer/producer.go b/producer/producer.go
index 2031549..5590f76 100644
--- a/producer/producer.go
+++ b/producer/producer.go
@@ -263,15 +263,12 @@ func (p *defaultProducer) sendOneWay(ctx context.Context, msg *primitive.Message
 
 func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue,
 	msg *primitive.Message) *remote.RemotingCommand {
-	if msg.Properties == nil {
-		msg.Properties = make(map[string]string, 0)
-	}
-	if !msg.Batch && msg.Properties[primitive.PropertyUniqueClientMessageIdKeyIndex] == "" {
-		msg.Properties[primitive.PropertyUniqueClientMessageIdKeyIndex] = primitive.CreateUniqID()
+	if !msg.Batch && msg.GetProperty(primitive.PropertyUniqueClientMessageIdKeyIndex) == "" {
+		msg.WithProperty(primitive.PropertyUniqueClientMessageIdKeyIndex, primitive.CreateUniqID())
 	}
 	sysFlag := 0
-	v, ok := msg.Properties[primitive.PropertyTransactionPrepared]
-	if ok {
+	v := msg.GetProperty(primitive.PropertyTransactionPrepared)
+	if v != "" {
 		tranMsg, err := strconv.ParseBool(v)
 		if err == nil && tranMsg {
 			sysFlag |= primitive.TransactionPreparedType
@@ -285,7 +282,7 @@ func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue,
 		SysFlag:        sysFlag,
 		BornTimestamp:  time.Now().UnixNano() / int64(time.Millisecond),
 		Flag:           msg.Flag,
-		Properties:     primitive.MarshalPropeties(msg.Properties),
+		Properties:     msg.MarshallProperties(),
 		ReconsumeTimes: 0,
 		UnitMode:       p.options.UnitMode,
 		Batch:          false,
@@ -380,8 +377,8 @@ func (tp *transactionProducer) checkTransactionState() {
 		switch callback := ch.(type) {
 		case internal.CheckTransactionStateCallback:
 			localTransactionState := tp.listener.CheckLocalTransaction(callback.Msg)
-			uniqueKey, existed := callback.Msg.Properties[primitive.PropertyUniqueClientMessageIdKeyIndex]
-			if !existed {
+			uniqueKey := callback.Msg.GetProperty(primitive.PropertyUniqueClientMessageIdKeyIndex)
+			if uniqueKey == "" {
 				uniqueKey = callback.Msg.MsgId
 			}
 			header := &internal.EndTransactionRequestHeader{
@@ -405,11 +402,8 @@ func (tp *transactionProducer) checkTransactionState() {
 }
 
 func (tp *transactionProducer) SendMessageInTransaction(ctx context.Context, msg *primitive.Message) (*primitive.TransactionSendResult, error) {
-	if msg.Properties == nil {
-		msg.Properties = make(map[string]string, 0)
-	}
-	msg.Properties[primitive.PropertyTransactionPrepared] = "true"
-	msg.Properties[primitive.PropertyProducerGroup] = tp.producer.options.GroupName
+	msg.WithProperty(primitive.PropertyTransactionPrepared, "true")
+	msg.WithProperty(primitive.PropertyProducerGroup, tp.producer.options.GroupName)
 
 	rsp, err := tp.producer.SendSync(ctx, msg)
 	if err != nil {
@@ -419,9 +413,9 @@ func (tp *transactionProducer) SendMessageInTransaction(ctx context.Context, msg
 	switch rsp.Status {
 	case primitive.SendOK:
 		if len(rsp.TransactionID) > 0 {
-			msg.Properties["__transactionId__"] = rsp.TransactionID
+			msg.WithProperty("__transactionId__", rsp.TransactionID)
 		}
-		transactionId := msg.Properties[primitive.PropertyUniqueClientMessageIdKeyIndex]
+		transactionId := msg.GetProperty(primitive.PropertyUniqueClientMessageIdKeyIndex)
 		if len(transactionId) > 0 {
 			msg.TransactionId = transactionId
 		}


Test patch:
diff --git a/primitive/result_test.go b/primitive/result_test.go
index 9324aff..00acdfb 100644
--- a/primitive/result_test.go
+++ b/primitive/result_test.go
@@ -25,13 +25,15 @@ import (
 )
 
 func TestProperties(t *testing.T) {
-	kv := map[string]string{
+	msg1 := NewMessage("test", nil)
+	msg1.properties = map[string]string{
 		"k1": "v1",
 		"k2": "v2",
 	}
-	str := MarshalPropeties(kv)
-	kv2 := unmarshalProperties([]byte(str))
-	assert.Equal(t, kv, kv2)
+	str := msg1.MarshallProperties()
+	msg2 := NewMessage("test", nil)
+	msg2.UnmarshalProperties([]byte(str))
+	assert.Equal(t, msg1.properties, msg2.properties)
 }
 
 func TestCreateMessageId(t *testing.T) {
diff --git a/producer/producer_test.go b/producer/producer_test.go
index 092dd9a..f26fa25 100644
--- a/producer/producer_test.go
+++ b/producer/producer_test.go
@@ -115,15 +115,15 @@ func TestSync(t *testing.T) {
 
 	ctx := context.Background()
 	msg := &primitive.Message{
-		Topic:      topic,
-		Body:       []byte("this is a message body"),
-		Properties: map[string]string{"key": "value"},
+		Topic: topic,
+		Body:  []byte("this is a message body"),
 		Queue: &primitive.MessageQueue{
 			Topic:      topic,
 			BrokerName: "aa",
 			QueueId:    0,
 		},
 	}
+	msg.WithProperty("key", "value")
 
 	expectedResp := &primitive.SendResult{
 		Status:      primitive.SendOK,
@@ -166,15 +166,15 @@ func TestASync(t *testing.T) {
 
 	ctx := context.Background()
 	msg := &primitive.Message{
-		Topic:      topic,
-		Body:       []byte("this is a message body"),
-		Properties: map[string]string{"key": "value"},
+		Topic: topic,
+		Body:  []byte("this is a message body"),
 		Queue: &primitive.MessageQueue{
 			Topic:      topic,
 			BrokerName: "aa",
 			QueueId:    0,
 		},
 	}
+	msg.WithProperty("key", "value")
 
 	expectedResp := &primitive.SendResult{
 		Status:      primitive.SendOK,
@@ -228,15 +228,15 @@ func TestOneway(t *testing.T) {
 
 	ctx := context.Background()
 	msg := &primitive.Message{
-		Topic:      topic,
-		Body:       []byte("this is a message body"),
-		Properties: map[string]string{"key": "value"},
+		Topic: topic,
+		Body:  []byte("this is a message body"),
 		Queue: &primitive.MessageQueue{
 			Topic:      topic,
 			BrokerName: "aa",
 			QueueId:    0,
 		},
 	}
+	msg.WithProperty("key", "value")
 
 	mockB4Send(p)