[rocketmq-client-go] branch native updated: Refactor directories & API (#82)

[hu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

huzongtang pushed a commit to branch native
in repository https://gitbox.apache.org/repos/asf/rocketmq-client-go.git


The following commit(s) were added to refs/heads/native by this push:
     new 3011efa  Refactor directories & API (#82)
3011efa is described below

commit 3011efa4508e453f14e0ef83647aa0e429e1cb35
Author: wenfeng <sx...@gmail.com>
AuthorDate: Tue Jul 2 23:40:15 2019 +0800

    Refactor directories & API (#82)
    
    * refactor directorie & apiss
    
    * fix ci failed
---
 api.go                                             |  54 ++++
 consumer/strategy.go                               | 130 ---------
 docs/Introduction.md                               |  12 +-
 examples/consumer/main.go                          |  16 +-
 examples/producer/main.go                          |   8 +-
 {consumer => internal/consumer}/consumer.go        | 301 ++++-----------------
 {consumer => internal/consumer}/consumer_test.go   |   0
 {consumer => internal/consumer}/offset_store.go    |  33 +--
 {consumer => internal/consumer}/process_queue.go   |  24 +-
 {consumer => internal/consumer}/pull_consumer.go   |  50 ++--
 .../consumer}/pull_consumer_test.go                |   0
 {consumer => internal/consumer}/push_consumer.go   | 103 +++----
 .../consumer}/push_consumer_test.go                |   0
 {consumer => internal/consumer}/statistics.go      |   0
 {kernel => internal/kernel}/client.go              |  83 +++---
 {kernel => internal/kernel}/client_test.go         |   0
 {kernel => internal/kernel}/constants.go           |   0
 internal/kernel/model.go                           |  80 ++++++
 {kernel => internal/kernel}/mq_version.go          |   0
 {kernel => internal/kernel}/perm.go                |   0
 {kernel => internal/kernel}/request.go             |   0
 {kernel => internal/kernel}/response.go            |   0
 {kernel => internal/kernel}/route.go               |  15 +-
 {kernel => internal/kernel}/route_test.go          |   0
 {kernel => internal/kernel}/transaction.go         |   0
 {kernel => internal/kernel}/validators.go          |   0
 {producer => internal/producer}/producer.go        |  29 +-
 {remote => internal/remote}/codec.go               |   0
 {remote => internal/remote}/codec_test.go          |   0
 {remote => internal/remote}/remote_client.go       |   0
 {remote => internal/remote}/remote_client_test.go  |   0
 {remote => internal/remote}/rpchook.go             |   0
 primitive/consume.go                               | 128 +++++++++
 {kernel => primitive}/message.go                   |  32 ++-
 primitive/options.go                               | 132 +++++++++
 kernel/model.go => primitive/result.go             |  86 +-----
 primitive/strategy.go                              | 117 ++++++++
 37 files changed, 760 insertions(+), 673 deletions(-)

diff --git a/api.go b/api.go
new file mode 100644
index 0000000..5043f95
--- /dev/null
+++ b/api.go
@@ -0,0 +1,54 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package rocketmq
+
+import (
+	"context"
+	"github.com/apache/rocketmq-client-go/primitive"
+)
+
+type Producer interface {
+	Start() error
+	Shutdown() error
+	SendSync(context.Context, ...*primitive.Message) (primitive.SendResult, error)
+	SendAsync(context.Context, func(primitive.SendResult), ...*primitive.Message) error
+	SendOneWay(context.Context, ...*primitive.Message) error
+}
+
+func NewProducer(opt primitive.ProducerOptions) (Producer, error) {
+	return nil, nil
+}
+
+type PushConsumer interface {
+	Start() error
+	Shutdown() error
+	Subscribe(topic string, selector primitive.MessageSelector,
+		f func(context.Context, ...*primitive.MessageExt) (primitive.ConsumeResult, error)) error
+	Unsubscribe(string) error
+}
+
+type PullConsumer interface {
+	Start() error
+	Shutdown() error
+	Pull(context.Context, string, primitive.MessageSelector, int) (primitive.PullResult, error)
+	PullFrom(context.Context, primitive.MessageQueue, int64, int) (primitive.PullResult, error)
+	// only update in memory
+	UpdateOffset(primitive.MessageQueue, int64) error
+	PersistOffset(context.Context) error
+	CurrentOffset(primitive.MessageQueue) int64
+}
diff --git a/consumer/strategy.go b/consumer/strategy.go
deleted file mode 100644
index 1cb6b2b..0000000
--- a/consumer/strategy.go
+++ /dev/null
@@ -1,130 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one or more
-contributor license agreements.  See the NOTICE file distributed with
-this work for additional information regarding copyright ownership.
-The ASF licenses this file to You under the Apache License, Version 2.0
-(the "License"); you may not use this file except in compliance with
-the License.  You may obtain a copy of the License at
-
-    http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing, software
-distributed under the License is distributed on an "AS IS" BASIS,
-WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-See the License for the specific language governing permissions and
-limitations under the License.
-*/
-
-package consumer
-
-import (
-	"github.com/apache/rocketmq-client-go/kernel"
-	"github.com/apache/rocketmq-client-go/rlog"
-	"github.com/apache/rocketmq-client-go/utils"
-)
-
-// Strategy Algorithm for message allocating between consumers
-type AllocateStrategy string
-
-const (
-	// An allocate strategy proxy for based on machine room nearside priority. An actual allocate strategy can be
-	// specified.
-	//
-	// If any consumer is alive in a machine room, the message queue of the broker which is deployed in the same machine
-	// should only be allocated to those. Otherwise, those message queues can be shared along all consumers since there are
-	// no alive consumer to monopolize them.
-	StrategyMachineNearby = AllocateStrategy("MachineNearby")
-
-	// Average Hashing queue algorithm
-	StrategyAveragely = AllocateStrategy("Averagely")
-
-	// Cycle average Hashing queue algorithm
-	StrategyAveragelyCircle = AllocateStrategy("AveragelyCircle")
-
-	// Use Message Queue specified
-	StrategyConfig = AllocateStrategy("Config")
-
-	// Computer room Hashing queue algorithm, such as Alipay logic room
-	StrategyMachineRoom = AllocateStrategy("MachineRoom")
-
-	// Consistent Hashing queue algorithm
-	StrategyConsistentHash = AllocateStrategy("ConsistentHash")
-)
-
-func allocateByAveragely(consumerGroup, currentCID string, mqAll []*kernel.MessageQueue,
-	cidAll []string) []*kernel.MessageQueue {
-	if currentCID == "" || utils.IsArrayEmpty(mqAll) || utils.IsArrayEmpty(cidAll) {
-		return nil
-	}
-	var (
-		find  bool
-		index int
-	)
-
-	for idx := range cidAll {
-		if cidAll[idx] == currentCID {
-			find = true
-			index = idx
-			break
-		}
-	}
-	if !find {
-		rlog.Infof("[BUG] ConsumerGroup=%s, ConsumerId=%s not in cidAll:%+v", consumerGroup, currentCID, cidAll)
-		return nil
-	}
-
-	mqSize := len(mqAll)
-	cidSize := len(cidAll)
-	mod := mqSize % cidSize
-
-	var averageSize int
-	if mqSize <= cidSize {
-		averageSize = 1
-	} else {
-		if mod > 0 && index < mod {
-			averageSize = mqSize/cidSize + 1
-		} else {
-			averageSize = mqSize / cidSize
-		}
-	}
-
-	var startIndex int
-	if mod > 0 && index < mod {
-		startIndex = index * averageSize
-	} else {
-		startIndex = index*averageSize + mod
-	}
-
-	num := utils.MinInt(averageSize, mqSize-startIndex)
-	result := make([]*kernel.MessageQueue, num)
-	for i := 0; i < num; i++ {
-		result[i] = mqAll[(startIndex+i)%mqSize]
-	}
-	return result
-}
-
-// TODO
-func allocateByMachineNearby(consumerGroup, currentCID string, mqAll []*kernel.MessageQueue,
-	cidAll []string) []*kernel.MessageQueue {
-	return allocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
-}
-
-func allocateByAveragelyCircle(consumerGroup, currentCID string, mqAll []*kernel.MessageQueue,
-	cidAll []string) []*kernel.MessageQueue {
-	return allocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
-}
-
-func allocateByConfig(consumerGroup, currentCID string, mqAll []*kernel.MessageQueue,
-	cidAll []string) []*kernel.MessageQueue {
-	return allocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
-}
-
-func allocateByMachineRoom(consumerGroup, currentCID string, mqAll []*kernel.MessageQueue,
-	cidAll []string) []*kernel.MessageQueue {
-	return allocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
-}
-
-func allocateByConsistentHash(consumerGroup, currentCID string, mqAll []*kernel.MessageQueue,
-	cidAll []string) []*kernel.MessageQueue {
-	return allocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
-}
diff --git a/docs/Introduction.md b/docs/Introduction.md
index 9bc22c7..0db6e7b 100644
--- a/docs/Introduction.md
+++ b/docs/Introduction.md
@@ -20,8 +20,8 @@ rlog.SetLogger(Logger)
 Producer interface {
 	Start() error
 	Shutdown() error
-	SendSync(context.Context, *kernel.Message) (*kernel.SendResult, error)
-	SendOneWay(context.Context, *kernel.Message) error
+	SendSync(context.Context, *primitive.Message) (*kernel.SendResult, error)
+	SendOneWay(context.Context, *primitive.Message) error
 }
 ```
 
@@ -42,7 +42,7 @@ err := p.Start()
 
 - send message with sync
 ```go
-result, err := p.SendSync(context.Background(), &kernel.Message{
+result, err := p.SendSync(context.Background(), &primitive.Message{
     Topic: "test",
     Body:  []byte("Hello RocketMQ Go Client!"),
 })
@@ -52,7 +52,7 @@ result, err := p.SendSync(context.Background(), &kernel.Message{
 
 - or send message with oneway
 ```go 
-err := p.SendOneWay(context.Background(), &kernel.Message{
+err := p.SendOneWay(context.Background(), &primitive.Message{
     Topic: "test",
     Body:  []byte("Hello RocketMQ Go Client!"),
 })
@@ -68,7 +68,7 @@ PushConsumer interface {
 	Start() error
 	Shutdown()
 	Subscribe(topic string, selector MessageSelector,
-		f func(*ConsumeMessageContext, []*kernel.MessageExt) (ConsumeResult, error)) error
+		f func(*ConsumeMessageContext, []*primitive.MessageExt) (ConsumeResult, error)) error
 }
 ```
 
@@ -85,7 +85,7 @@ c := consumer.NewPushConsumer("testGroup", consumer.ConsumerOption{
 - Subscribe a topic(only support one topic now), and define your consuming function
 ```go
 err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx *consumer.ConsumeMessageContext,
-    msgs []*kernel.MessageExt) (consumer.ConsumeResult, error) {
+    msgs []*primitive.MessageExt) (consumer.ConsumeResult, error) {
     fmt.Println(msgs)
     return consumer.ConsumeSuccess, nil
 })
diff --git a/examples/consumer/main.go b/examples/consumer/main.go
index f660d43..65b0ba3 100644
--- a/examples/consumer/main.go
+++ b/examples/consumer/main.go
@@ -19,22 +19,22 @@ package main
 
 import (
 	"fmt"
-	"github.com/apache/rocketmq-client-go/consumer"
-	"github.com/apache/rocketmq-client-go/kernel"
+	"github.com/apache/rocketmq-client-go/internal/consumer"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"os"
 	"time"
 )
 
 func main() {
-	c, _ := consumer.NewPushConsumer("testGroup", consumer.ConsumerOption{
+	c, _ := consumer.NewPushConsumer("testGroup", primitive.ConsumerOption{
 		NameServerAddr: "127.0.0.1:9876",
-		ConsumerModel:  consumer.Clustering,
-		FromWhere:      consumer.ConsumeFromFirstOffset,
+		ConsumerModel:  primitive.Clustering,
+		FromWhere:      primitive.ConsumeFromFirstOffset,
 	})
-	err := c.Subscribe("test", consumer.MessageSelector{}, func(ctx *consumer.ConsumeMessageContext,
-		msgs []*kernel.MessageExt) (consumer.ConsumeResult, error) {
+	err := c.Subscribe("test", primitive.MessageSelector{}, func(ctx *consumer.ConsumeMessageContext,
+		msgs []*primitive.MessageExt) (primitive.ConsumeResult, error) {
 		fmt.Println(msgs)
-		return consumer.ConsumeSuccess, nil
+		return primitive.ConsumeSuccess, nil
 	})
 	if err != nil {
 		fmt.Println(err.Error())
diff --git a/examples/producer/main.go b/examples/producer/main.go
index 9cc626b..c3e338d 100644
--- a/examples/producer/main.go
+++ b/examples/producer/main.go
@@ -20,13 +20,13 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/kernel"
-	"github.com/apache/rocketmq-client-go/producer"
+	"github.com/apache/rocketmq-client-go/internal/producer"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"os"
 )
 
 func main() {
-	opt := producer.ProducerOptions{
+	opt := primitive.ProducerOptions{
 		NameServerAddr:           "127.0.0.1:9876",
 		RetryTimesWhenSendFailed: 2,
 	}
@@ -37,7 +37,7 @@ func main() {
 		os.Exit(1)
 	}
 	for i := 0; i < 1000; i++ {
-		res, err := p.SendSync(context.Background(), &kernel.Message{
+		res, err := p.SendSync(context.Background(), &primitive.Message{
 			Topic: "test",
 			Body:  []byte("Hello RocketMQ Go Client!"),
 		})
diff --git a/consumer/consumer.go b/internal/consumer/consumer.go
similarity index 66%
rename from consumer/consumer.go
rename to internal/consumer/consumer.go
index 80e4fb5..08ae664 100644
--- a/consumer/consumer.go
+++ b/internal/consumer/consumer.go
@@ -20,8 +20,9 @@ package consumer
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/kernel"
-	"github.com/apache/rocketmq-client-go/remote"
+	"github.com/apache/rocketmq-client-go/internal/kernel"
+	"github.com/apache/rocketmq-client-go/internal/remote"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 	"github.com/apache/rocketmq-client-go/utils"
 	"github.com/tidwall/gjson"
@@ -53,135 +54,18 @@ const (
 	_PersistConsumerOffsetInterval = 5 * time.Second
 )
 
-// Message model defines the way how messages are delivered to each consumer clients.
-// </p>
-//
-// RocketMQ supports two message models: clustering and broadcasting. If clustering is set, consumer clients with
-// the same {@link #consumerGroup} would only consume shards of the messages subscribed, which achieves load
-// balances; Conversely, if the broadcasting is set, each consumer client will consume all subscribed messages
-// separately.
-// </p>
-//
-// This field defaults to clustering.
-type MessageModel int
-
-const (
-	BroadCasting MessageModel = iota
-	Clustering
-)
-
-func (mode MessageModel) String() string {
-	switch mode {
-	case BroadCasting:
-		return "BroadCasting"
-	case Clustering:
-		return "Clustering"
-	default:
-		return "Unknown"
-	}
-}
-
-// Consuming point on consumer booting.
-// </p>
-//
-// There are three consuming points:
-// <ul>
-// <li>
-// <code>CONSUME_FROM_LAST_OFFSET</code>: consumer clients pick up where it stopped previously.
-// If it were a newly booting up consumer client, according aging of the consumer group, there are two
-// cases:
-// <ol>
-// <li>
-// if the consumer group is created so recently that the earliest message being subscribed has yet
-// expired, which means the consumer group represents a lately launched business, consuming will
-// start from the very beginning;
-// </li>
-// <li>
-// if the earliest message being subscribed has expired, consuming will start from the latest
-// messages, meaning messages born prior to the booting timestamp would be ignored.
-// </li>
-// </ol>
-// </li>
-// <li>
-// <code>CONSUME_FROM_FIRST_OFFSET</code>: Consumer client will start from earliest messages available.
-// </li>
-// <li>
-// <code>CONSUME_FROM_TIMESTAMP</code>: Consumer client will start from specified timestamp, which means
-// messages born prior to {@link #consumeTimestamp} will be ignored
-// </li>
-// </ul>
-type ConsumeFromWhere int
-
-const (
-	ConsumeFromLastOffset ConsumeFromWhere = iota
-	ConsumeFromFirstOffset
-	ConsumeFromTimestamp
-)
-
 type ConsumeType string
 
 const (
 	_PullConsume = ConsumeType("pull")
 	_PushConsume = ConsumeType("push")
-)
-
-type ExpressionType string
-
-const (
-	/**
-	 * <ul>
-	 * Keywords:
-	 * <li>{@code AND, OR, NOT, BETWEEN, IN, TRUE, FALSE, IS, NULL}</li>
-	 * </ul>
-	 * <p/>
-	 * <ul>
-	 * Data type:
-	 * <li>Boolean, like: TRUE, FALSE</li>
-	 * <li>String, like: 'abc'</li>
-	 * <li>Decimal, like: 123</li>
-	 * <li>Float number, like: 3.1415</li>
-	 * </ul>
-	 * <p/>
-	 * <ul>
-	 * Grammar:
-	 * <li>{@code AND, OR}</li>
-	 * <li>{@code >, >=, <, <=, =}</li>
-	 * <li>{@code BETWEEN A AND B}, equals to {@code >=A AND <=B}</li>
-	 * <li>{@code NOT BETWEEN A AND B}, equals to {@code >B OR <A}</li>
-	 * <li>{@code IN ('a', 'b')}, equals to {@code ='a' OR ='b'}, this operation only support String type.</li>
-	 * <li>{@code IS NULL}, {@code IS NOT NULL}, check parameter whether is null, or not.</li>
-	 * <li>{@code =TRUE}, {@code =FALSE}, check parameter whether is true, or false.</li>
-	 * </ul>
-	 * <p/>
-	 * <p>
-	 * Example:
-	 * (a > 10 AND a < 100) OR (b IS NOT NULL AND b=TRUE)
-	 * </p>
-	 */
-	SQL92 = ExpressionType("SQL92")
-
-	/**
-	 * Only support or operation such as
-	 * "tag1 || tag2 || tag3", <br>
-	 * If null or * expression, meaning subscribe all.
-	 */
-	TAG = ExpressionType("TAG")
-)
-
-func IsTagType(exp string) bool {
-	if exp == "" || exp == "TAG" {
-		return true
-	}
-	return false
-}
 
-const (
 	_SubAll = "*"
 )
 
 type PullRequest struct {
 	consumerGroup string
-	mq            *kernel.MessageQueue
+	mq            *primitive.MessageQueue
 	pq            *processQueue
 	nextOffset    int64
 	lockedFirst   bool
@@ -192,83 +76,6 @@ func (pr *PullRequest) String() string {
 		pr.consumerGroup, pr.mq.Topic, pr.mq.QueueId)
 }
 
-type ConsumerOption struct {
-	kernel.ClientOption
-	NameServerAddr string
-
-	/**
-	 * Backtracking consumption time with second precision. Time format is
-	 * 20131223171201<br>
-	 * Implying Seventeen twelve and 01 seconds on December 23, 2013 year<br>
-	 * Default backtracking consumption time Half an hour ago.
-	 */
-	ConsumeTimestamp string
-
-	// The socket timeout in milliseconds
-	ConsumerPullTimeout time.Duration
-
-	// Concurrently max span offset.it has no effect on sequential consumption
-	ConsumeConcurrentlyMaxSpan int
-
-	// Flow control threshold on queue level, each message queue will cache at most 1000 messages by default,
-	// Consider the {PullBatchSize}, the instantaneous value may exceed the limit
-	PullThresholdForQueue int64
-
-	// Limit the cached message size on queue level, each message queue will cache at most 100 MiB messages by default,
-	// Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit
-	//
-	// The size of a message only measured by message body, so it's not accurate
-	PullThresholdSizeForQueue int
-
-	// Flow control threshold on topic level, default value is -1(Unlimited)
-	//
-	// The value of {@code pullThresholdForQueue} will be overwrote and calculated based on
-	// {@code pullThresholdForTopic} if it is't unlimited
-	//
-	// For example, if the value of pullThresholdForTopic is 1000 and 10 message queues are assigned to this consumer,
-	// then pullThresholdForQueue will be set to 100
-	PullThresholdForTopic int
-
-	// Limit the cached message size on topic level, default value is -1 MiB(Unlimited)
-	//
-	// The value of {@code pullThresholdSizeForQueue} will be overwrote and calculated based on
-	// {@code pullThresholdSizeForTopic} if it is't unlimited
-	//
-	// For example, if the value of pullThresholdSizeForTopic is 1000 MiB and 10 message queues are
-	// assigned to this consumer, then pullThresholdSizeForQueue will be set to 100 MiB
-	PullThresholdSizeForTopic int
-
-	// Message pull Interval
-	PullInterval time.Duration
-
-	// Batch consumption size
-	ConsumeMessageBatchMaxSize int
-
-	// Batch pull size
-	PullBatchSize int32
-
-	// Whether update subscription relationship when every pull
-	PostSubscriptionWhenPull bool
-
-	// Max re-consume times. -1 means 16 times.
-	//
-	// If messages are re-consumed more than {@link #maxReconsumeTimes} before success, it's be directed to a deletion
-	// queue waiting.
-	MaxReconsumeTimes int
-
-	// Suspending pulling time for cases requiring slow pulling like flow-control scenario.
-	SuspendCurrentQueueTimeMillis time.Duration
-
-	// Maximum amount of time a message may block the consuming thread.
-	ConsumeTimeout time.Duration
-
-	ConsumerModel  MessageModel
-	Strategy       AllocateStrategy
-	ConsumeOrderly bool
-	FromWhere      ConsumeFromWhere
-	// TODO traceDispatcher
-}
-
 // TODO hook
 type defaultConsumer struct {
 	/**
@@ -279,25 +86,25 @@ type defaultConsumer struct {
 	 * See <a href="http://rocketmq.apache.org/docs/core-concept/">here</a> for further discussion.
 	 */
 	consumerGroup  string
-	model          MessageModel
-	allocate       func(string, string, []*kernel.MessageQueue, []string) []*kernel.MessageQueue
+	model          primitive.MessageModel
+	allocate       func(string, string, []*primitive.MessageQueue, []string) []*primitive.MessageQueue
 	unitMode       bool
 	consumeOrderly bool
-	fromWhere      ConsumeFromWhere
+	fromWhere      primitive.ConsumeFromWhere
 
 	cType     ConsumeType
 	client    *kernel.RMQClient
-	mqChanged func(topic string, mqAll, mqDivided []*kernel.MessageQueue)
+	mqChanged func(topic string, mqAll, mqDivided []*primitive.MessageQueue)
 	state     kernel.ServiceState
 	pause     bool
 	once      sync.Once
-	option    ConsumerOption
-	// key: int, hash(*kernel.MessageQueue)
+	option    primitive.ConsumerOption
+	// key: int, hash(*primitive.MessageQueue)
 	// value: *processQueue
 	processQueueTable sync.Map
 
 	// key: topic(string)
-	// value: map[int]*kernel.MessageQueue
+	// value: map[int]*primitive.MessageQueue
 	topicSubscribeInfoTable sync.Map
 
 	// key: topic
@@ -314,19 +121,19 @@ func (dc *defaultConsumer) persistConsumerOffset() {
 		rlog.Errorf("consumer state error: %s", err.Error())
 		return
 	}
-	mqs := make([]*kernel.MessageQueue, 0)
+	mqs := make([]*primitive.MessageQueue, 0)
 	dc.processQueueTable.Range(func(key, value interface{}) bool {
-		mqs = append(mqs, key.(*kernel.MessageQueue))
+		mqs = append(mqs, key.(*primitive.MessageQueue))
 		return true
 	})
 	dc.storage.persist(mqs)
 }
 
-func (dc *defaultConsumer) updateTopicSubscribeInfo(topic string, mqs []*kernel.MessageQueue) {
+func (dc *defaultConsumer) updateTopicSubscribeInfo(topic string, mqs []*primitive.MessageQueue) {
 	_, exist := dc.subscriptionDataTable.Load(topic)
 	// does subscribe, if true, replace it
 	if exist {
-		mqSet := make(map[int]*kernel.MessageQueue, 0)
+		mqSet := make(map[int]*primitive.MessageQueue, 0)
 		for idx := range mqs {
 			mq := mqs[idx]
 			mqSet[mq.HashCode()] = mq
@@ -355,23 +162,23 @@ func (dc *defaultConsumer) doBalance() {
 			rlog.Warnf("do balance of group: %s, but topic: %s does not exist.", dc.consumerGroup, topic)
 			return true
 		}
-		mqs := v.([]*kernel.MessageQueue)
+		mqs := v.([]*primitive.MessageQueue)
 		switch dc.model {
-		case BroadCasting:
+		case primitive.BroadCasting:
 			changed := dc.updateProcessQueueTable(topic, mqs)
 			if changed {
 				dc.mqChanged(topic, mqs, mqs)
 				rlog.Infof("messageQueueChanged, Group: %s, Topic: %s, MessageQueues: %v",
 					dc.consumerGroup, topic, mqs)
 			}
-		case Clustering:
+		case primitive.Clustering:
 			cidAll := dc.findConsumerList(topic)
 			if cidAll == nil {
 				rlog.Warnf("do balance for Group: %s, Topic: %s get consumer id list failed",
 					dc.consumerGroup, topic)
 				return true
 			}
-			mqAll := make([]*kernel.MessageQueue, len(mqs))
+			mqAll := make([]*primitive.MessageQueue, len(mqs))
 			copy(mqAll, mqs)
 			sort.Strings(cidAll)
 			sort.SliceStable(mqAll, func(i, j int) bool {
@@ -390,9 +197,9 @@ func (dc *defaultConsumer) doBalance() {
 			changed := dc.updateProcessQueueTable(topic, allocateResult)
 			if changed {
 				dc.mqChanged(topic, mqAll, allocateResult)
-				rlog.Infof("do balance result changed, allocateMessageQueueStrategyName=%s, group=%s, "+
+				rlog.Infof("do balance result changed, group=%s, "+
 					"topic=%s, clientId=%s, mqAllSize=%d, cidAllSize=%d, rebalanceResultSize=%d, "+
-					"rebalanceResultSet=%v", string(dc.option.Strategy), dc.consumerGroup, topic, dc.client.ClientID(), len(mqAll),
+					"rebalanceResultSet=%v", dc.consumerGroup, topic, dc.client.ClientID(), len(mqAll),
 					len(cidAll), len(allocateResult), allocateResult)
 
 			}
@@ -418,12 +225,12 @@ func (dc *defaultConsumer) makeSureStateOK() error {
 }
 
 type lockBatchRequestBody struct {
-	ConsumerGroup string                 `json:"consumerGroup"`
-	ClientId      string                 `json:"clientId"`
-	MQs           []*kernel.MessageQueue `json:"mqSet"`
+	ConsumerGroup string                    `json:"consumerGroup"`
+	ClientId      string                    `json:"clientId"`
+	MQs           []*primitive.MessageQueue `json:"mqSet"`
 }
 
-func (dc *defaultConsumer) lock(mq *kernel.MessageQueue) bool {
+func (dc *defaultConsumer) lock(mq *primitive.MessageQueue) bool {
 	brokerResult := kernel.FindBrokerAddressInSubscribe(mq.BrokerName, kernel.MasterId, true)
 
 	if brokerResult == nil {
@@ -433,7 +240,7 @@ func (dc *defaultConsumer) lock(mq *kernel.MessageQueue) bool {
 	body := &lockBatchRequestBody{
 		ConsumerGroup: dc.consumerGroup,
 		ClientId:      dc.client.ClientID(),
-		MQs:           []*kernel.MessageQueue{mq},
+		MQs:           []*primitive.MessageQueue{mq},
 	}
 	lockedMQ := dc.doLock(brokerResult.BrokerAddr, body)
 	var lockOK bool
@@ -453,7 +260,7 @@ func (dc *defaultConsumer) lock(mq *kernel.MessageQueue) bool {
 	return lockOK
 }
 
-func (dc *defaultConsumer) unlock(mq *kernel.MessageQueue, oneway bool) {
+func (dc *defaultConsumer) unlock(mq *primitive.MessageQueue, oneway bool) {
 	brokerResult := kernel.FindBrokerAddressInSubscribe(mq.BrokerName, kernel.MasterId, true)
 
 	if brokerResult == nil {
@@ -463,14 +270,14 @@ func (dc *defaultConsumer) unlock(mq *kernel.MessageQueue, oneway bool) {
 	body := &lockBatchRequestBody{
 		ConsumerGroup: dc.consumerGroup,
 		ClientId:      dc.client.ClientID(),
-		MQs:           []*kernel.MessageQueue{mq},
+		MQs:           []*primitive.MessageQueue{mq},
 	}
 	dc.doUnlock(brokerResult.BrokerAddr, body, oneway)
 	rlog.Warnf("unlock messageQueue. group:%s, clientId:%s, mq:%s",
 		dc.consumerGroup, dc.client.ClientID(), mq.String())
 }
 
-func (dc *defaultConsumer) lockAll(mq kernel.MessageQueue) {
+func (dc *defaultConsumer) lockAll(mq primitive.MessageQueue) {
 	mqMapSet := dc.buildProcessQueueTableByBrokerName()
 	for broker, mqs := range mqMapSet {
 		if len(mqs) == 0 {
@@ -539,7 +346,7 @@ func (dc *defaultConsumer) unlockAll(oneway bool) {
 	}
 }
 
-func (dc *defaultConsumer) doLock(addr string, body *lockBatchRequestBody) []kernel.MessageQueue {
+func (dc *defaultConsumer) doLock(addr string, body *lockBatchRequestBody) []primitive.MessageQueue {
 	data, _ := json.Marshal(body)
 	request := remote.NewRemotingCommand(kernel.ReqLockBatchMQ, nil, data)
 	response, err := dc.client.InvokeSync(addr, request, 1*time.Second)
@@ -548,7 +355,7 @@ func (dc *defaultConsumer) doLock(addr string, body *lockBatchRequestBody) []ker
 		return nil
 	}
 	lockOKMQSet := struct {
-		MQs []kernel.MessageQueue `json:"lockOKMQSet"`
+		MQs []primitive.MessageQueue `json:"lockOKMQSet"`
 	}{}
 	err = json.Unmarshal(response.Body, &lockOKMQSet)
 	if err != nil {
@@ -577,14 +384,14 @@ func (dc *defaultConsumer) doUnlock(addr string, body *lockBatchRequestBody, one
 	}
 }
 
-func (dc *defaultConsumer) buildProcessQueueTableByBrokerName() map[string][]*kernel.MessageQueue {
-	result := make(map[string][]*kernel.MessageQueue, 0)
+func (dc *defaultConsumer) buildProcessQueueTableByBrokerName() map[string][]*primitive.MessageQueue {
+	result := make(map[string][]*primitive.MessageQueue, 0)
 
 	dc.processQueueTable.Range(func(key, value interface{}) bool {
-		mq := key.(*kernel.MessageQueue)
+		mq := key.(*primitive.MessageQueue)
 		mqs, exist := result[mq.BrokerName]
 		if !exist {
-			mqs = make([]*kernel.MessageQueue, 0)
+			mqs = make([]*primitive.MessageQueue, 0)
 		}
 		mqs = append(mqs, mq)
 		result[mq.BrokerName] = mqs
@@ -595,15 +402,15 @@ func (dc *defaultConsumer) buildProcessQueueTableByBrokerName() map[string][]*ke
 }
 
 // TODO 问题不少 需要再好好对一下
-func (dc *defaultConsumer) updateProcessQueueTable(topic string, mqs []*kernel.MessageQueue) bool {
+func (dc *defaultConsumer) updateProcessQueueTable(topic string, mqs []*primitive.MessageQueue) bool {
 	var changed bool
-	mqSet := make(map[*kernel.MessageQueue]bool)
+	mqSet := make(map[*primitive.MessageQueue]bool)
 	for idx := range mqs {
 		mqSet[mqs[idx]] = true
 	}
 	// TODO
 	dc.processQueueTable.Range(func(key, value interface{}) bool {
-		mq := key.(*kernel.MessageQueue)
+		mq := key.(*primitive.MessageQueue)
 		pq := value.(*processQueue)
 		if mq.Topic == topic {
 			if !mqSet[mq] {
@@ -666,13 +473,13 @@ func (dc *defaultConsumer) updateProcessQueueTable(topic string, mqs []*kernel.M
 	return changed
 }
 
-func (dc *defaultConsumer) removeUnnecessaryMessageQueue(mq *kernel.MessageQueue, pq *processQueue) bool {
-	dc.storage.persist([]*kernel.MessageQueue{mq})
+func (dc *defaultConsumer) removeUnnecessaryMessageQueue(mq *primitive.MessageQueue, pq *processQueue) bool {
+	dc.storage.persist([]*primitive.MessageQueue{mq})
 	dc.storage.remove(mq)
 	return true
 }
 
-func (dc *defaultConsumer) computePullFromWhere(mq *kernel.MessageQueue) int64 {
+func (dc *defaultConsumer) computePullFromWhere(mq *primitive.MessageQueue) int64 {
 	if dc.cType == _PullConsume {
 		return 0
 	}
@@ -682,7 +489,7 @@ func (dc *defaultConsumer) computePullFromWhere(mq *kernel.MessageQueue) int64 {
 		result = lastOffset
 	} else {
 		switch dc.fromWhere {
-		case ConsumeFromLastOffset:
+		case primitive.ConsumeFromLastOffset:
 			if lastOffset == -1 {
 				if strings.HasPrefix(mq.Topic, kernel.RetryGroupTopicPrefix) {
 					lastOffset = 0
@@ -697,11 +504,11 @@ func (dc *defaultConsumer) computePullFromWhere(mq *kernel.MessageQueue) int64 {
 			} else {
 				result = -1
 			}
-		case ConsumeFromFirstOffset:
+		case primitive.ConsumeFromFirstOffset:
 			if lastOffset == -1 {
 				result = 0
 			}
-		case ConsumeFromTimestamp:
+		case primitive.ConsumeFromTimestamp:
 			if lastOffset == -1 {
 				if strings.HasPrefix(mq.Topic, kernel.RetryGroupTopicPrefix) {
 					lastOffset, err := dc.queryMaxOffset(mq)
@@ -760,12 +567,12 @@ func (dc *defaultConsumer) findConsumerList(topic string) []string {
 	return nil
 }
 
-func (dc *defaultConsumer) sendBack(msg *kernel.MessageExt, level int) error {
+func (dc *defaultConsumer) sendBack(msg *primitive.MessageExt, level int) error {
 	return nil
 }
 
 // QueryMaxOffset with specific queueId and topic
-func (dc *defaultConsumer) queryMaxOffset(mq *kernel.MessageQueue) (int64, error) {
+func (dc *defaultConsumer) queryMaxOffset(mq *primitive.MessageQueue) (int64, error) {
 	brokerAddr := kernel.FindBrokerAddrByName(mq.BrokerName)
 	if brokerAddr == "" {
 		kernel.UpdateTopicRouteInfo(mq.Topic)
@@ -790,7 +597,7 @@ func (dc *defaultConsumer) queryMaxOffset(mq *kernel.MessageQueue) (int64, error
 }
 
 // SearchOffsetByTimestamp with specific queueId and topic
-func (dc *defaultConsumer) searchOffsetByTimestamp(mq *kernel.MessageQueue, timestamp int64) (int64, error) {
+func (dc *defaultConsumer) searchOffsetByTimestamp(mq *primitive.MessageQueue, timestamp int64) (int64, error) {
 	brokerAddr := kernel.FindBrokerAddrByName(mq.BrokerName)
 	if brokerAddr == "" {
 		kernel.UpdateTopicRouteInfo(mq.Topic)
@@ -815,19 +622,19 @@ func (dc *defaultConsumer) searchOffsetByTimestamp(mq *kernel.MessageQueue, time
 	return strconv.ParseInt(response.ExtFields["offset"], 10, 64)
 }
 
-func buildSubscriptionData(topic string, selector MessageSelector) *kernel.SubscriptionData {
+func buildSubscriptionData(topic string, selector primitive.MessageSelector) *kernel.SubscriptionData {
 	subData := &kernel.SubscriptionData{
 		Topic:     topic,
 		SubString: selector.Expression,
 		ExpType:   string(selector.Type),
 	}
 
-	if selector.Type != "" && selector.Type != TAG {
+	if selector.Type != "" && selector.Type != primitive.TAG {
 		return subData
 	}
 
 	if selector.Expression == "" || selector.Expression == _SubAll {
-		subData.ExpType = string(TAG)
+		subData.ExpType = string(primitive.TAG)
 		subData.SubString = _SubAll
 	} else {
 		tags := strings.Split(selector.Expression, "\\|\\|")
@@ -847,7 +654,7 @@ func buildSubscriptionData(topic string, selector MessageSelector) *kernel.Subsc
 	return subData
 }
 
-func getNextQueueOf(topic string) *kernel.MessageQueue {
+func getNextQueueOf(topic string) *primitive.MessageQueue {
 	queues, err := kernel.FetchSubscribeMessageQueues(topic)
 	if err != nil && len(queues) > 0 {
 		rlog.Error(err.Error())
@@ -890,7 +697,7 @@ func clearCommitOffsetFlag(sysFlag int32) int32 {
 	return sysFlag & (^0x1 << 0)
 }
 
-func tryFindBroker(mq *kernel.MessageQueue) *kernel.FindBrokerResult {
+func tryFindBroker(mq *primitive.MessageQueue) *kernel.FindBrokerResult {
 	result := kernel.FindBrokerAddressInSubscribe(mq.BrokerName, recalculatePullFromWhichNode(mq), false)
 
 	if result == nil {
@@ -903,11 +710,11 @@ var (
 	pullFromWhichNodeTable sync.Map
 )
 
-func updatePullFromWhichNode(mq *kernel.MessageQueue, brokerId int64) {
+func updatePullFromWhichNode(mq *primitive.MessageQueue, brokerId int64) {
 	pullFromWhichNodeTable.Store(mq.HashCode(), brokerId)
 }
 
-func recalculatePullFromWhichNode(mq *kernel.MessageQueue) int64 {
+func recalculatePullFromWhichNode(mq *primitive.MessageQueue) int64 {
 	v, exist := pullFromWhichNodeTable.Load(mq.HashCode())
 	if exist {
 		return v.(int64)
diff --git a/consumer/consumer_test.go b/internal/consumer/consumer_test.go
similarity index 100%
rename from consumer/consumer_test.go
rename to internal/consumer/consumer_test.go
diff --git a/consumer/offset_store.go b/internal/consumer/offset_store.go
similarity index 86%
rename from consumer/offset_store.go
rename to internal/consumer/offset_store.go
index 0edf828..702ba0d 100644
--- a/consumer/offset_store.go
+++ b/internal/consumer/offset_store.go
@@ -20,8 +20,9 @@ package consumer
 import (
 	"encoding/json"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/kernel"
-	"github.com/apache/rocketmq-client-go/remote"
+	"github.com/apache/rocketmq-client-go/internal/kernel"
+	"github.com/apache/rocketmq-client-go/internal/remote"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 	"github.com/apache/rocketmq-client-go/utils"
 	"os"
@@ -50,10 +51,10 @@ func init() {
 }
 
 type OffsetStore interface {
-	persist(mqs []*kernel.MessageQueue)
-	remove(mq *kernel.MessageQueue)
-	read(mq *kernel.MessageQueue, t readType) int64
-	update(mq *kernel.MessageQueue, offset int64, increaseOnly bool)
+	persist(mqs []*primitive.MessageQueue)
+	remove(mq *primitive.MessageQueue)
+	read(mq *primitive.MessageQueue, t readType) int64
+	update(mq *primitive.MessageQueue, offset int64, increaseOnly bool)
 }
 
 type localFileOffsetStore struct {
@@ -108,7 +109,7 @@ func (local *localFileOffsetStore) load() {
 	}
 }
 
-func (local *localFileOffsetStore) read(mq *kernel.MessageQueue, t readType) int64 {
+func (local *localFileOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
 	if t == _ReadFromMemory || t == _ReadMemoryThenStore {
 		off := readFromMemory(local.OffsetTable, mq)
 		if off >= 0 || (off == -1 && t == _ReadFromMemory) {
@@ -119,7 +120,7 @@ func (local *localFileOffsetStore) read(mq *kernel.MessageQueue, t readType) int
 	return readFromMemory(local.OffsetTable, mq)
 }
 
-func (local *localFileOffsetStore) update(mq *kernel.MessageQueue, offset int64, increaseOnly bool) {
+func (local *localFileOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
 	rlog.Debugf("update offset: %s to %d", mq, offset)
 	localOffset, exist := local.OffsetTable[mq.Topic]
 	if !exist {
@@ -143,7 +144,7 @@ func (local *localFileOffsetStore) update(mq *kernel.MessageQueue, offset int64,
 	}
 }
 
-func (local *localFileOffsetStore) persist(mqs []*kernel.MessageQueue) {
+func (local *localFileOffsetStore) persist(mqs []*primitive.MessageQueue) {
 	if len(mqs) == 0 {
 		return
 	}
@@ -171,7 +172,7 @@ func (local *localFileOffsetStore) persist(mqs []*kernel.MessageQueue) {
 	utils.CheckError(fmt.Sprintf("persist offset to %s", local.path), utils.WriteToFile(local.path, data))
 }
 
-func (local *localFileOffsetStore) remove(mq *kernel.MessageQueue) {
+func (local *localFileOffsetStore) remove(mq *primitive.MessageQueue) {
 	// nothing to do
 }
 
@@ -190,7 +191,7 @@ func NewRemoteOffsetStore(group string, client *kernel.RMQClient) OffsetStore {
 	}
 }
 
-func (r *remoteBrokerOffsetStore) persist(mqs []*kernel.MessageQueue) {
+func (r *remoteBrokerOffsetStore) persist(mqs []*primitive.MessageQueue) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 	if len(mqs) == 0 {
@@ -217,7 +218,7 @@ func (r *remoteBrokerOffsetStore) persist(mqs []*kernel.MessageQueue) {
 	}
 }
 
-func (r *remoteBrokerOffsetStore) remove(mq *kernel.MessageQueue) {
+func (r *remoteBrokerOffsetStore) remove(mq *primitive.MessageQueue) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 	if mq == nil {
@@ -231,7 +232,7 @@ func (r *remoteBrokerOffsetStore) remove(mq *kernel.MessageQueue) {
 	delete(offset, mq.QueueId)
 }
 
-func (r *remoteBrokerOffsetStore) read(mq *kernel.MessageQueue, t readType) int64 {
+func (r *remoteBrokerOffsetStore) read(mq *primitive.MessageQueue, t readType) int64 {
 	r.mutex.RLock()
 	if t == _ReadFromMemory || t == _ReadMemoryThenStore {
 		off := readFromMemory(r.OffsetTable, mq)
@@ -251,7 +252,7 @@ func (r *remoteBrokerOffsetStore) read(mq *kernel.MessageQueue, t readType) int6
 	return off
 }
 
-func (r *remoteBrokerOffsetStore) update(mq *kernel.MessageQueue, offset int64, increaseOnly bool) {
+func (r *remoteBrokerOffsetStore) update(mq *primitive.MessageQueue, offset int64, increaseOnly bool) {
 	rlog.Debugf("update offset: %s to %d", mq, offset)
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
@@ -278,7 +279,7 @@ func (r *remoteBrokerOffsetStore) update(mq *kernel.MessageQueue, offset int64,
 	}
 }
 
-func (r *remoteBrokerOffsetStore) fetchConsumeOffsetFromBroker(group string, mq *kernel.MessageQueue) (int64, error) {
+func (r *remoteBrokerOffsetStore) fetchConsumeOffsetFromBroker(group string, mq *primitive.MessageQueue) (int64, error) {
 	broker := kernel.FindBrokerAddrByName(mq.BrokerName)
 	if broker == "" {
 		kernel.UpdateTopicRouteInfo(mq.Topic)
@@ -330,7 +331,7 @@ func (r *remoteBrokerOffsetStore) updateConsumeOffsetToBroker(group, topic strin
 	return r.client.InvokeOneWay(broker, cmd, 5*time.Second)
 }
 
-func readFromMemory(table map[string]map[int]*queueOffset, mq *kernel.MessageQueue) int64 {
+func readFromMemory(table map[string]map[int]*queueOffset, mq *primitive.MessageQueue) int64 {
 	localOffset, exist := table[mq.Topic]
 	if !exist {
 		return -1
diff --git a/consumer/process_queue.go b/internal/consumer/process_queue.go
similarity index 87%
rename from consumer/process_queue.go
rename to internal/consumer/process_queue.go
index 77e84fa..2ec865c 100644
--- a/consumer/process_queue.go
+++ b/internal/consumer/process_queue.go
@@ -18,7 +18,7 @@ limitations under the License.
 package consumer
 
 import (
-	"github.com/apache/rocketmq-client-go/kernel"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 	"github.com/emirpasic/gods/maps/treemap"
 	"github.com/emirpasic/gods/utils"
@@ -51,7 +51,7 @@ type processQueue struct {
 	consuming                  bool
 	msgAccCnt                  int64
 	lockConsume                sync.Mutex
-	msgCh                      chan []*kernel.MessageExt
+	msgCh                      chan []*primitive.MessageExt
 }
 
 func newProcessQueue() *processQueue {
@@ -60,12 +60,12 @@ func newProcessQueue() *processQueue {
 		lastPullTime:    time.Now(),
 		lastConsumeTime: time.Now(),
 		lastLockTime:    time.Now(),
-		msgCh:           make(chan []*kernel.MessageExt, 32),
+		msgCh:           make(chan []*primitive.MessageExt, 32),
 	}
 	return pq
 }
 
-func (pq *processQueue) putMessage(messages ...*kernel.MessageExt) {
+func (pq *processQueue) putMessage(messages ...*primitive.MessageExt) {
 	if messages == nil || len(messages) == 0 {
 		return
 	}
@@ -92,7 +92,7 @@ func (pq *processQueue) putMessage(messages ...*kernel.MessageExt) {
 	}
 
 	msg := messages[len(messages)-1]
-	maxOffset, err := strconv.ParseInt(msg.Properties[kernel.PropertyMaxOffset], 10, 64)
+	maxOffset, err := strconv.ParseInt(msg.Properties[primitive.PropertyMaxOffset], 10, 64)
 	if err != nil {
 		acc := maxOffset - msg.QueueOffset
 		if acc > 0 {
@@ -101,7 +101,7 @@ func (pq *processQueue) putMessage(messages ...*kernel.MessageExt) {
 	}
 }
 
-func (pq *processQueue) removeMessage(messages ...*kernel.MessageExt) int64 {
+func (pq *processQueue) removeMessage(messages ...*primitive.MessageExt) int64 {
 	result := int64(-1)
 	pq.mutex.Lock()
 	pq.lastConsumeTime = time.Now()
@@ -152,8 +152,8 @@ func (pq *processQueue) cleanExpiredMsg(consumer defaultConsumer) {
 			return
 		}
 		_, firstValue := pq.msgCache.Min()
-		msg := firstValue.(*kernel.MessageExt)
-		startTime := msg.Properties[kernel.PropertyConsumeStartTime]
+		msg := firstValue.(*primitive.MessageExt)
+		startTime := msg.Properties[primitive.PropertyConsumeStartTime]
 		if startTime != "" {
 			st, err := strconv.ParseInt(startTime, 10, 64)
 			if err != nil {
@@ -187,15 +187,15 @@ func (pq *processQueue) getMaxSpan() int {
 	return int(lastKey.(int64) - firstKey.(int64))
 }
 
-func (pq *processQueue) getMessages() []*kernel.MessageExt {
+func (pq *processQueue) getMessages() []*primitive.MessageExt {
 	return <-pq.msgCh
 }
 
-func (pq *processQueue) takeMessages(number int) []*kernel.MessageExt {
+func (pq *processQueue) takeMessages(number int) []*primitive.MessageExt {
 	for pq.msgCache.Empty() {
 		time.Sleep(10 * time.Millisecond)
 	}
-	result := make([]*kernel.MessageExt, number)
+	result := make([]*primitive.MessageExt, number)
 	i := 0
 	pq.mutex.Lock()
 	for ; i < number; i++ {
@@ -203,7 +203,7 @@ func (pq *processQueue) takeMessages(number int) []*kernel.MessageExt {
 		if v == nil {
 			break
 		}
-		result[i] = v.(*kernel.MessageExt)
+		result[i] = v.(*primitive.MessageExt)
 		pq.msgCache.Remove(k)
 	}
 	pq.mutex.Unlock()
diff --git a/consumer/pull_consumer.go b/internal/consumer/pull_consumer.go
similarity index 70%
rename from consumer/pull_consumer.go
rename to internal/consumer/pull_consumer.go
index 1b4f950..9250771 100644
--- a/consumer/pull_consumer.go
+++ b/internal/consumer/pull_consumer.go
@@ -21,27 +21,23 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/kernel"
+	"github.com/apache/rocketmq-client-go/internal/kernel"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"strconv"
 	"sync"
 )
 
-type MessageSelector struct {
-	Type       ExpressionType
-	Expression string
-}
-
 type PullConsumer interface {
 	Start()
 	Shutdown()
-	Pull(ctx context.Context, topic string, selector MessageSelector, numbers int) (*kernel.PullResult, error)
+	Pull(ctx context.Context, topic string, selector primitive.MessageSelector, numbers int) (*primitive.PullResult, error)
 }
 
 var (
 	queueCounterTable sync.Map
 )
 
-func NewConsumer(config ConsumerOption) *defaultPullConsumer {
+func NewConsumer(config primitive.ConsumerOption) *defaultPullConsumer {
 	return &defaultPullConsumer{
 		option: config,
 	}
@@ -49,10 +45,10 @@ func NewConsumer(config ConsumerOption) *defaultPullConsumer {
 
 type defaultPullConsumer struct {
 	state     kernel.ServiceState
-	option    ConsumerOption
+	option    primitive.ConsumerOption
 	client    *kernel.RMQClient
 	GroupName string
-	Model     MessageModel
+	Model     primitive.MessageModel
 	UnitMode  bool
 }
 
@@ -60,7 +56,7 @@ func (c *defaultPullConsumer) Start() {
 	c.state = kernel.StateRunning
 }
 
-func (c *defaultPullConsumer) Pull(ctx context.Context, topic string, selector MessageSelector, numbers int) (*kernel.PullResult, error) {
+func (c *defaultPullConsumer) Pull(ctx context.Context, topic string, selector primitive.MessageSelector, numbers int) (*primitive.PullResult, error) {
 	mq := getNextQueueOf(topic)
 	if mq == nil {
 		return nil, fmt.Errorf("prepard to pull topic: %s, but no queue is founded", topic)
@@ -78,22 +74,22 @@ func (c *defaultPullConsumer) Pull(ctx context.Context, topic string, selector M
 }
 
 // SubscribeWithChan ack manually
-func (c *defaultPullConsumer) SubscribeWithChan(topic, selector MessageSelector) (chan *kernel.Message, error) {
+func (c *defaultPullConsumer) SubscribeWithChan(topic, selector primitive.MessageSelector) (chan *primitive.Message, error) {
 	return nil, nil
 }
 
 // SubscribeWithFunc ack automatic
-func (c *defaultPullConsumer) SubscribeWithFunc(topic, selector MessageSelector,
-	f func(msg *kernel.Message) ConsumeResult) error {
+func (c *defaultPullConsumer) SubscribeWithFunc(topic, selector primitive.MessageSelector,
+	f func(msg *primitive.Message) primitive.ConsumeResult) error {
 	return nil
 }
 
-func (c *defaultPullConsumer) ACK(msg *kernel.Message, result ConsumeResult) {
+func (c *defaultPullConsumer) ACK(msg *primitive.Message, result primitive.ConsumeResult) {
 
 }
 
-func (c *defaultPullConsumer) pull(ctx context.Context, mq *kernel.MessageQueue, data *kernel.SubscriptionData,
-	offset int64, numbers int) (*kernel.PullResult, error) {
+func (c *defaultPullConsumer) pull(ctx context.Context, mq *primitive.MessageQueue, data *kernel.SubscriptionData,
+	offset int64, numbers int) (*primitive.PullResult, error) {
 	err := c.makeSureStateOK()
 	if err != nil {
 		return nil, err
@@ -117,7 +113,7 @@ func (c *defaultPullConsumer) pull(ctx context.Context, mq *kernel.MessageQueue,
 		return nil, fmt.Errorf("the broker %s does not exist", mq.BrokerName)
 	}
 
-	if (data.ExpType == string(TAG)) && brokerResult.BrokerVersion < kernel.V4_1_0 {
+	if (data.ExpType == string(primitive.TAG)) && brokerResult.BrokerVersion < kernel.V4_1_0 {
 		return nil, fmt.Errorf("the broker [%s, %v] does not upgrade to support for filter message by %v",
 			mq.BrokerName, brokerResult.BrokerVersion, data.ExpType)
 	}
@@ -140,7 +136,7 @@ func (c *defaultPullConsumer) pull(ctx context.Context, mq *kernel.MessageQueue,
 		ExpressionType:       string(data.ExpType),
 	}
 
-	if data.ExpType == string(TAG) {
+	if data.ExpType == string(primitive.TAG) {
 		pullRequest.SubVersion = 0
 	} else {
 		pullRequest.SubVersion = data.SubVersion
@@ -161,18 +157,18 @@ func (c *defaultPullConsumer) subscriptionAutomatically(topic string) {
 	// TODO
 }
 
-func (c *defaultPullConsumer) nextOffsetOf(queue *kernel.MessageQueue) int64 {
+func (c *defaultPullConsumer) nextOffsetOf(queue *primitive.MessageQueue) int64 {
 	return 0
 }
 
-func processPullResult(mq *kernel.MessageQueue, result *kernel.PullResult, data *kernel.SubscriptionData) {
+func processPullResult(mq *primitive.MessageQueue, result *primitive.PullResult, data *kernel.SubscriptionData) {
 	updatePullFromWhichNode(mq, result.SuggestWhichBrokerId)
 	switch result.Status {
-	case kernel.PullFound:
+	case primitive.PullFound:
 		msgs := result.GetMessageExts()
 		msgListFilterAgain := msgs
 		if len(data.Tags) > 0 && data.ClassFilterMode {
-			msgListFilterAgain = make([]*kernel.MessageExt, len(msgs))
+			msgListFilterAgain = make([]*primitive.MessageExt, len(msgs))
 			for _, msg := range msgs {
 				_, exist := data.Tags[msg.GetTags()]
 				if exist {
@@ -184,13 +180,13 @@ func processPullResult(mq *kernel.MessageQueue, result *kernel.PullResult, data
 		// TODO hook
 
 		for _, msg := range msgListFilterAgain {
-			traFlag, _ := strconv.ParseBool(msg.Properties[kernel.PropertyTransactionPrepared])
+			traFlag, _ := strconv.ParseBool(msg.Properties[primitive.PropertyTransactionPrepared])
 			if traFlag {
-				msg.TransactionId = msg.Properties[kernel.PropertyUniqueClientMessageIdKeyIndex]
+				msg.TransactionId = msg.Properties[primitive.PropertyUniqueClientMessageIdKeyIndex]
 			}
 
-			msg.Properties[kernel.PropertyMinOffset] = strconv.FormatInt(result.MinOffset, 10)
-			msg.Properties[kernel.PropertyMaxOffset] = strconv.FormatInt(result.MaxOffset, 10)
+			msg.Properties[primitive.PropertyMinOffset] = strconv.FormatInt(result.MinOffset, 10)
+			msg.Properties[primitive.PropertyMaxOffset] = strconv.FormatInt(result.MaxOffset, 10)
 		}
 
 		result.SetMessageExts(msgListFilterAgain)
diff --git a/consumer/pull_consumer_test.go b/internal/consumer/pull_consumer_test.go
similarity index 100%
rename from consumer/pull_consumer_test.go
rename to internal/consumer/pull_consumer_test.go
diff --git a/consumer/push_consumer.go b/internal/consumer/push_consumer.go
similarity index 88%
rename from consumer/push_consumer.go
rename to internal/consumer/push_consumer.go
index 1cb5efd..b3d1e5d 100644
--- a/consumer/push_consumer.go
+++ b/internal/consumer/push_consumer.go
@@ -21,7 +21,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/kernel"
+	"github.com/apache/rocketmq-client-go/internal/kernel"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 	"github.com/apache/rocketmq-client-go/utils"
 	"math"
@@ -38,31 +39,28 @@ import (
 // See quick start/Consumer in the example module for a typical usage.
 //
 // <strong>Thread Safety:</strong> After initialization, the instance can be regarded as thread-safe.
-type ConsumeResult int
 
 const (
-	Mb                           = 1024 * 1024
-	ConsumeSuccess ConsumeResult = iota
-	ConsumeRetryLater
+	Mb = 1024 * 1024
 )
 
 type PushConsumer interface {
 	Start() error
 	Shutdown()
-	Subscribe(topic string, selector MessageSelector,
-		f func(*ConsumeMessageContext, []*kernel.MessageExt) (ConsumeResult, error)) error
+	Subscribe(topic string, selector primitive.MessageSelector,
+		f func(*ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult, error)) error
 }
 
 type pushConsumer struct {
 	*defaultConsumer
 	queueFlowControlTimes        int
 	queueMaxSpanFlowControlTimes int
-	consume                      func(*ConsumeMessageContext, []*kernel.MessageExt) (ConsumeResult, error)
-	submitToConsume              func(*processQueue, *kernel.MessageQueue)
+	consume                      func(*ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult, error)
+	submitToConsume              func(*processQueue, *primitive.MessageQueue)
 	subscribedTopic              map[string]string
 }
 
-func NewPushConsumer(consumerGroup string, opt ConsumerOption) (PushConsumer, error) {
+func NewPushConsumer(consumerGroup string, opt primitive.ConsumerOption) (PushConsumer, error) {
 	if err := utils.VerifyIP(opt.NameServerAddr); err != nil {
 		return nil, err
 	}
@@ -86,23 +84,10 @@ func NewPushConsumer(consumerGroup string, opt ConsumerOption) (PushConsumer, er
 		option:         opt,
 	}
 
-	switch opt.Strategy {
-	case StrategyAveragely:
-		dc.allocate = allocateByAveragely
-	case StrategyAveragelyCircle:
-		dc.allocate = allocateByAveragelyCircle
-	case StrategyConfig:
-		dc.allocate = allocateByConfig
-	case StrategyConsistentHash:
-		dc.allocate = allocateByConsistentHash
-	case StrategyMachineNearby:
-		dc.allocate = allocateByMachineNearby
-	case StrategyMachineRoom:
-		dc.allocate = allocateByMachineRoom
-	default:
-		dc.allocate = allocateByAveragely
+	if opt.Strategy == nil {
+		opt.Strategy = primitive.AllocateByAveragely
 	}
-
+	dc.allocate = opt.Strategy
 	p := &pushConsumer{
 		defaultConsumer: dc,
 		subscribedTopic: make(map[string]string, 0),
@@ -124,15 +109,15 @@ func (pc *pushConsumer) Start() error {
 		pc.state = kernel.StateStartFailed
 		pc.validate()
 
-		if pc.model == Clustering {
+		if pc.model == primitive.Clustering {
 			// set retry topic
 			retryTopic := kernel.GetRetryTopic(pc.consumerGroup)
 			pc.subscriptionDataTable.Store(retryTopic, buildSubscriptionData(retryTopic,
-				MessageSelector{TAG, _SubAll}))
+				primitive.MessageSelector{primitive.TAG, _SubAll}))
 		}
 
 		pc.client = kernel.GetOrNewRocketMQClient(pc.option.ClientOption)
-		if pc.model == Clustering {
+		if pc.model == primitive.Clustering {
 			pc.option.ChangeInstanceNameToPID()
 			pc.storage = NewRemoteOffsetStore(pc.consumerGroup, pc.client)
 		} else {
@@ -177,8 +162,8 @@ func (pc *pushConsumer) Start() error {
 
 func (pc *pushConsumer) Shutdown() {}
 
-func (pc *pushConsumer) Subscribe(topic string, selector MessageSelector,
-	f func(*ConsumeMessageContext, []*kernel.MessageExt) (ConsumeResult, error)) error {
+func (pc *pushConsumer) Subscribe(topic string, selector primitive.MessageSelector,
+	f func(*ConsumeMessageContext, []*primitive.MessageExt) (primitive.ConsumeResult, error)) error {
 	if pc.state != kernel.StateCreateJust {
 		return errors.New("subscribe topic only started before")
 	}
@@ -197,7 +182,7 @@ func (pc *pushConsumer) PersistConsumerOffset() {
 	pc.defaultConsumer.persistConsumerOffset()
 }
 
-func (pc *pushConsumer) UpdateTopicSubscribeInfo(topic string, mqs []*kernel.MessageQueue) {
+func (pc *pushConsumer) UpdateTopicSubscribeInfo(topic string, mqs []*primitive.MessageQueue) {
 	pc.defaultConsumer.updateTopicSubscribeInfo(topic, mqs)
 }
 
@@ -213,7 +198,7 @@ func (pc *pushConsumer) IsUnitMode() bool {
 	return pc.unitMode
 }
 
-func (pc *pushConsumer) messageQueueChanged(topic string, mqAll, mqDivided []*kernel.MessageQueue) {
+func (pc *pushConsumer) messageQueueChanged(topic string, mqAll, mqDivided []*primitive.MessageQueue) {
 	// TODO
 }
 
@@ -399,7 +384,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 			subExpression      string
 		)
 
-		if pc.model == Clustering {
+		if pc.model == primitive.Clustering {
 			commitOffsetValue = pc.storage.read(request.mq, _ReadFromMemory)
 			if commitOffsetValue > 0 {
 				commitOffsetEnable = true
@@ -423,7 +408,7 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 			SysFlag:        sysFlag,
 			CommitOffset:   commitOffsetValue,
 			SubExpression:  _SubAll,
-			ExpressionType: string(TAG), // TODO
+			ExpressionType: string(primitive.TAG), // TODO
 		}
 		//
 		//if data.ExpType == string(TAG) {
@@ -445,14 +430,14 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 			goto NEXT
 		}
 
-		if result.Status == kernel.PullBrokerTimeout {
+		if result.Status == primitive.PullBrokerTimeout {
 			rlog.Warnf("pull broker: %s timeout", brokerResult.BrokerAddr)
 			sleepTime = _PullDelayTimeWhenError
 			goto NEXT
 		}
 
 		switch result.Status {
-		case kernel.PullFound:
+		case primitive.PullFound:
 			rlog.Debugf("Topic: %s, QueueId: %d found messages: %d", request.mq.Topic, request.mq.QueueId,
 				len(result.GetMessageExts()))
 			prevRequestOffset := request.nextOffset
@@ -472,19 +457,19 @@ func (pc *pushConsumer) pullMessage(request *PullRequest) {
 				rlog.Warnf("[BUG] pull message result maybe data wrong, [nextBeginOffset=%d, "+
 					"firstMsgOffset=%d, prevRequestOffset=%d]", result.NextBeginOffset, firstMsgOffset, prevRequestOffset)
 			}
-		case kernel.PullNoNewMsg:
+		case primitive.PullNoNewMsg:
 			rlog.Debugf("Topic: %s, QueueId: %d no more msg, next offset: %d", request.mq.Topic, request.mq.QueueId, result.NextBeginOffset)
-		case kernel.PullNoMsgMatched:
+		case primitive.PullNoMsgMatched:
 			request.nextOffset = result.NextBeginOffset
 			pc.correctTagsOffset(request)
-		case kernel.PullOffsetIllegal:
+		case primitive.PullOffsetIllegal:
 			rlog.Warnf("the pull request offset illegal, {} {}", request.String(), result.String())
 			request.nextOffset = result.NextBeginOffset
 			pq.dropped = true
 			go func() {
 				time.Sleep(10 * time.Second)
 				pc.storage.update(request.mq, request.nextOffset, false)
-				pc.storage.persist([]*kernel.MessageQueue{request.mq})
+				pc.storage.persist([]*primitive.MessageQueue{request.mq})
 				pc.storage.remove(request.mq)
 				rlog.Warnf("fix the pull request offset: %s", request.String())
 			}()
@@ -499,7 +484,7 @@ func (pc *pushConsumer) correctTagsOffset(pr *PullRequest) {
 	// TODO
 }
 
-func (pc *pushConsumer) sendMessageBack(ctx *ConsumeMessageContext, msg *kernel.MessageExt) bool {
+func (pc *pushConsumer) sendMessageBack(ctx *ConsumeMessageContext, msg *primitive.MessageExt) bool {
 	return true
 }
 
@@ -514,7 +499,7 @@ func (pc *pushConsumer) resume() {
 	rlog.Infof("resume consumer: %s", pc.consumerGroup)
 }
 
-func (pc *pushConsumer) resetOffset(topic string, table map[kernel.MessageQueue]int64) {
+func (pc *pushConsumer) resetOffset(topic string, table map[primitive.MessageQueue]int64) {
 	//topic := cmd.ExtFields["topic"]
 	//group := cmd.ExtFields["group"]
 	//if topic == "" || group == "" {
@@ -529,7 +514,7 @@ func (pc *pushConsumer) resetOffset(topic string, table map[kernel.MessageQueue]
 	//rlog.Infof("invoke reset offset operation from broker. brokerAddr=%s, topic=%s, group=%s, timestamp=%v",
 	//	from, topic, group, t)
 	//
-	//offsetTable := make(map[kernel.MessageQueue]int64, 0)
+	//offsetTable := make(map[primitive.MessageQueue]int64, 0)
 	//err = json.Unmarshal(cmd.Body, &offsetTable)
 	//if err != nil {
 	//	rlog.Warnf("received reset offset command from: %s, but parse offset table: %s", err.Error())
@@ -541,7 +526,7 @@ func (pc *pushConsumer) resetOffset(topic string, table map[kernel.MessageQueue]
 	//	return
 	//}
 
-	set := make(map[int]*kernel.MessageQueue, 0)
+	set := make(map[int]*primitive.MessageQueue, 0)
 	for k := range table {
 		set[k.HashCode()] = &k
 	}
@@ -559,7 +544,7 @@ func (pc *pushConsumer) resetOffset(topic string, table map[kernel.MessageQueue]
 	if !exist {
 		return
 	}
-	queuesOfTopic := v.(map[int]*kernel.MessageQueue)
+	queuesOfTopic := v.(map[int]*primitive.MessageQueue)
 	for k := range queuesOfTopic {
 		q := set[k]
 		if q != nil {
@@ -575,9 +560,9 @@ func (pc *pushConsumer) resetOffset(topic string, table map[kernel.MessageQueue]
 	}
 }
 
-func (pc *pushConsumer) removeUnnecessaryMessageQueue(mq *kernel.MessageQueue, pq *processQueue) bool {
+func (pc *pushConsumer) removeUnnecessaryMessageQueue(mq *primitive.MessageQueue, pq *processQueue) bool {
 	pc.defaultConsumer.removeUnnecessaryMessageQueue(mq, pq)
-	if !pc.consumeOrderly || Clustering != pc.model {
+	if !pc.consumeOrderly || primitive.Clustering != pc.model {
 		return true
 	}
 	// TODO orderly
@@ -586,21 +571,21 @@ func (pc *pushConsumer) removeUnnecessaryMessageQueue(mq *kernel.MessageQueue, p
 
 type ConsumeMessageContext struct {
 	consumerGroup string
-	msgs          []*kernel.MessageExt
-	mq            *kernel.MessageQueue
+	msgs          []*primitive.MessageExt
+	mq            *primitive.MessageQueue
 	success       bool
 	status        string
 	// mqTractContext
 	properties map[string]string
 }
 
-func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *kernel.MessageQueue) {
+func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *primitive.MessageQueue) {
 	msgs := pq.getMessages()
 	if msgs == nil {
 		return
 	}
 	for count := 0; count < len(msgs); count++ {
-		var subMsgs []*kernel.MessageExt
+		var subMsgs []*primitive.MessageExt
 		if count+pc.option.ConsumeMessageBatchMaxSize > len(msgs) {
 			subMsgs = msgs[count:]
 			count = len(msgs)
@@ -626,11 +611,11 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *kernel.Mes
 			for idx := range subMsgs {
 				msg := subMsgs[idx]
 				if msg.Properties != nil {
-					retryTopic := msg.Properties[kernel.PropertyRetryTopic]
+					retryTopic := msg.Properties[primitive.PropertyRetryTopic]
 					if retryTopic == "" && groupTopic == msg.Topic {
 						msg.Topic = retryTopic
 					}
-					subMsgs[idx].Properties[kernel.PropertyConsumeStartTime] = strconv.FormatInt(
+					subMsgs[idx].Properties[primitive.PropertyConsumeStartTime] = strconv.FormatInt(
 						beginTime.UnixNano()/int64(time.Millisecond), 10)
 				}
 			}
@@ -640,7 +625,7 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *kernel.Mes
 				ctx.properties["ConsumeContextType"] = "EXCEPTION"
 			} else if consumeRT >= pc.option.ConsumeTimeout {
 				ctx.properties["ConsumeContextType"] = "TIMEOUT"
-			} else if result == ConsumeSuccess {
+			} else if result == primitive.ConsumeSuccess {
 				ctx.properties["ConsumeContextType"] = "SUCCESS"
 			} else {
 				ctx.properties["ConsumeContextType"] = "RECONSUME_LATER"
@@ -650,12 +635,12 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *kernel.Mes
 			increaseConsumeRT(pc.consumerGroup, mq.Topic, consumeRT)
 
 			if !pq.dropped {
-				msgBackFailed := make([]*kernel.MessageExt, 0)
-				if result == ConsumeSuccess {
+				msgBackFailed := make([]*primitive.MessageExt, 0)
+				if result == primitive.ConsumeSuccess {
 					increaseConsumeOKTPS(pc.consumerGroup, mq.Topic, len(subMsgs))
 				} else {
 					increaseConsumeFailedTPS(pc.consumerGroup, mq.Topic, len(subMsgs))
-					if pc.model == BroadCasting {
+					if pc.model == primitive.BroadCasting {
 						for i := 0; i < len(msgs); i++ {
 							rlog.Warnf("BROADCASTING, the message=%s consume failed, drop it, {}", subMsgs[i])
 						}
@@ -688,5 +673,5 @@ func (pc *pushConsumer) consumeMessageCurrently(pq *processQueue, mq *kernel.Mes
 	}
 }
 
-func (pc *pushConsumer) consumeMessageOrderly(pq *processQueue, mq *kernel.MessageQueue) {
+func (pc *pushConsumer) consumeMessageOrderly(pq *processQueue, mq *primitive.MessageQueue) {
 }
diff --git a/consumer/push_consumer_test.go b/internal/consumer/push_consumer_test.go
similarity index 100%
rename from consumer/push_consumer_test.go
rename to internal/consumer/push_consumer_test.go
diff --git a/consumer/statistics.go b/internal/consumer/statistics.go
similarity index 100%
rename from consumer/statistics.go
rename to internal/consumer/statistics.go
diff --git a/kernel/client.go b/internal/kernel/client.go
similarity index 86%
rename from kernel/client.go
rename to internal/kernel/client.go
index 9dbecc6..78de681 100644
--- a/kernel/client.go
+++ b/internal/kernel/client.go
@@ -22,7 +22,8 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/remote"
+	"github.com/apache/rocketmq-client-go/internal/remote"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 	"os"
 	"strconv"
@@ -65,28 +66,6 @@ func init() {
 	}
 }
 
-type ClientOption struct {
-	NameServerAddr    string
-	ClientIP          string
-	InstanceName      string
-	UnitMode          bool
-	UnitName          string
-	VIPChannelEnabled bool
-	UseTLS            bool
-}
-
-func (opt *ClientOption) ChangeInstanceNameToPID() {
-	if opt.InstanceName == "DEFAULT" {
-		opt.InstanceName = strconv.Itoa(os.Getegid())
-	}
-}
-
-func (opt *ClientOption) String() string {
-	return fmt.Sprintf("ClientOption [ClientIP=%s, InstanceName=%s, "+
-		"UnitMode=%v, UnitName=%s, VIPChannelEnabled=%v, UseTLS=%v]", opt.ClientIP,
-		opt.InstanceName, opt.UnitMode, opt.UnitName, opt.VIPChannelEnabled, opt.UseTLS)
-}
-
 type InnerProducer interface {
 	PublishTopicList() []string
 	UpdateTopicPublishInfo(topic string, info *TopicPublishInfo)
@@ -97,7 +76,7 @@ type InnerProducer interface {
 
 type InnerConsumer interface {
 	PersistConsumerOffset()
-	UpdateTopicSubscribeInfo(topic string, mqs []*MessageQueue)
+	UpdateTopicSubscribeInfo(topic string, mqs []*primitive.MessageQueue)
 	IsSubscribeTopicNeedUpdate(topic string) bool
 	SubscriptionDataList() []*SubscriptionData
 	Rebalance()
@@ -105,7 +84,7 @@ type InnerConsumer interface {
 }
 
 type RMQClient struct {
-	option ClientOption
+	option primitive.ClientOption
 	// group -> InnerProducer
 	producerMap sync.Map
 
@@ -119,7 +98,7 @@ type RMQClient struct {
 
 var clientMap sync.Map
 
-func GetOrNewRocketMQClient(option ClientOption) *RMQClient {
+func GetOrNewRocketMQClient(option primitive.ClientOption) *RMQClient {
 	client := &RMQClient{
 		option:       option,
 		remoteClient: remote.NewRemotingClient(),
@@ -298,12 +277,12 @@ func (c *RMQClient) UpdateTopicRouteInfo() {
 
 // SendMessageAsync send message with batch by async
 func (c *RMQClient) SendMessageAsync(ctx context.Context, brokerAddrs, brokerName string, request *SendMessageRequest,
-	msgs []*Message, f func(result *SendResult)) error {
+	msgs []*primitive.Message, f func(result *primitive.SendResult)) error {
 	return nil
 }
 
 func (c *RMQClient) SendMessageOneWay(ctx context.Context, brokerAddrs string, request *SendMessageRequest,
-	msgs []*Message) (*SendResult, error) {
+	msgs []*primitive.Message) (*primitive.SendResult, error) {
 	cmd := remote.NewRemotingCommand(ReqSendBatchMessage, request, encodeMessages(msgs))
 	err := c.remoteClient.InvokeOneWay(brokerAddrs, cmd, 3*time.Second)
 	if err != nil {
@@ -312,28 +291,28 @@ func (c *RMQClient) SendMessageOneWay(ctx context.Context, brokerAddrs string, r
 	return nil, err
 }
 
-func (c *RMQClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingCommand, msgs ...*Message) *SendResult {
-	var status SendStatus
+func (c *RMQClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingCommand, msgs ...*primitive.Message) *primitive.SendResult {
+	var status primitive.SendStatus
 	switch cmd.Code {
 	case ResFlushDiskTimeout:
-		status = SendFlushDiskTimeout
+		status = primitive.SendFlushDiskTimeout
 	case ResFlushSlaveTimeout:
-		status = SendFlushSlaveTimeout
+		status = primitive.SendFlushSlaveTimeout
 	case ResSlaveNotAvailable:
-		status = SendSlaveNotAvailable
+		status = primitive.SendSlaveNotAvailable
 	case ResSuccess:
-		status = SendOK
+		status = primitive.SendOK
 	default:
 		// TODO process unknown code
 	}
 
 	msgIDs := make([]string, 0)
 	for i := 0; i < len(msgs); i++ {
-		msgIDs = append(msgIDs, msgs[i].Properties[PropertyUniqueClientMessageIdKeyIndex])
+		msgIDs = append(msgIDs, msgs[i].Properties[primitive.PropertyUniqueClientMessageIdKeyIndex])
 	}
 
-	regionId := cmd.ExtFields[PropertyMsgRegion]
-	trace := cmd.ExtFields[PropertyTraceSwitch]
+	regionId := cmd.ExtFields[primitive.PropertyMsgRegion]
+	trace := cmd.ExtFields[primitive.PropertyTraceSwitch]
 
 	if regionId == "" {
 		regionId = defaultTraceRegionID
@@ -341,11 +320,11 @@ func (c *RMQClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingC
 
 	qId, _ := strconv.Atoi(cmd.ExtFields["queueId"])
 	off, _ := strconv.ParseInt(cmd.ExtFields["queueOffset"], 10, 64)
-	return &SendResult{
+	return &primitive.SendResult{
 		Status:      status,
 		MsgID:       cmd.ExtFields["msgId"],
 		OffsetMsgID: cmd.ExtFields["msgId"],
-		MessageQueue: &MessageQueue{
+		MessageQueue: &primitive.MessageQueue{
 			Topic:      msgs[0].Topic,
 			BrokerName: brokerName,
 			QueueId:    qId,
@@ -358,7 +337,7 @@ func (c *RMQClient) ProcessSendResponse(brokerName string, cmd *remote.RemotingC
 }
 
 // PullMessage with sync
-func (c *RMQClient) PullMessage(ctx context.Context, brokerAddrs string, request *PullMessageRequest) (*PullResult, error) {
+func (c *RMQClient) PullMessage(ctx context.Context, brokerAddrs string, request *PullMessageRequest) (*primitive.PullResult, error) {
 	cmd := remote.NewRemotingCommand(ReqPullMessage, request, nil)
 	res, err := c.remoteClient.InvokeSync(brokerAddrs, cmd, 3*time.Second)
 	if err != nil {
@@ -368,17 +347,17 @@ func (c *RMQClient) PullMessage(ctx context.Context, brokerAddrs string, request
 	return c.processPullResponse(res)
 }
 
-func (c *RMQClient) processPullResponse(response *remote.RemotingCommand) (*PullResult, error) {
-	pullResult := &PullResult{}
+func (c *RMQClient) processPullResponse(response *remote.RemotingCommand) (*primitive.PullResult, error) {
+	pullResult := &primitive.PullResult{}
 	switch response.Code {
 	case ResSuccess:
-		pullResult.Status = PullFound
+		pullResult.Status = primitive.PullFound
 	case ResPullNotFound:
-		pullResult.Status = PullNoNewMsg
+		pullResult.Status = primitive.PullNoNewMsg
 	case ResPullRetryImmediately:
-		pullResult.Status = PullNoMsgMatched
+		pullResult.Status = primitive.PullNoMsgMatched
 	case ResPullOffsetMoved:
-		pullResult.Status = PullOffsetIllegal
+		pullResult.Status = primitive.PullOffsetIllegal
 	default:
 		return nil, fmt.Errorf("unknown Response Code: %d, remark: %s", response.Code, response.Remark)
 	}
@@ -403,13 +382,13 @@ func (c *RMQClient) processPullResponse(response *remote.RemotingCommand) (*Pull
 		pullResult.SuggestWhichBrokerId, _ = strconv.ParseInt(v, 10, 64)
 	}
 
-	pullResult.messageExts = decodeMessage(response.Body)
+	//pullResult.messageExts = decodeMessage(response.Body) TODO parse in top
 
 	return pullResult, nil
 }
 
 // PullMessageAsync pull message async
-func (c *RMQClient) PullMessageAsync(ctx context.Context, brokerAddrs string, request *PullMessageRequest, f func(result *PullResult)) error {
+func (c *RMQClient) PullMessageAsync(ctx context.Context, brokerAddrs string, request *PullMessageRequest, f func(result *primitive.PullResult)) error {
 	return nil
 }
 
@@ -501,13 +480,13 @@ func (c *RMQClient) isNeedUpdateSubscribeInfo(topic string) bool {
 	return result
 }
 
-func routeData2SubscribeInfo(topic string, data *TopicRouteData) []*MessageQueue {
-	list := make([]*MessageQueue, 0)
+func routeData2SubscribeInfo(topic string, data *TopicRouteData) []*primitive.MessageQueue {
+	list := make([]*primitive.MessageQueue, 0)
 	for idx := range data.QueueDataList {
 		qd := data.QueueDataList[idx]
 		if queueIsReadable(qd.Perm) {
 			for i := 0; i < qd.ReadQueueNums; i++ {
-				list = append(list, &MessageQueue{
+				list = append(list, &primitive.MessageQueue{
 					Topic:      topic,
 					BrokerName: qd.BrokerName,
 					QueueId:    i,
@@ -518,7 +497,7 @@ func routeData2SubscribeInfo(topic string, data *TopicRouteData) []*MessageQueue
 	return list
 }
 
-func encodeMessages(message []*Message) []byte {
+func encodeMessages(message []*primitive.Message) []byte {
 	var buffer bytes.Buffer
 	index := 0
 	for index < len(message) {
diff --git a/kernel/client_test.go b/internal/kernel/client_test.go
similarity index 100%
rename from kernel/client_test.go
rename to internal/kernel/client_test.go
diff --git a/kernel/constants.go b/internal/kernel/constants.go
similarity index 100%
rename from kernel/constants.go
rename to internal/kernel/constants.go
diff --git a/internal/kernel/model.go b/internal/kernel/model.go
new file mode 100644
index 0000000..b3ef62e
--- /dev/null
+++ b/internal/kernel/model.go
@@ -0,0 +1,80 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kernel
+
+import (
+	"encoding/json"
+	"github.com/apache/rocketmq-client-go/rlog"
+)
+
+type FindBrokerResult struct {
+	BrokerAddr    string
+	Slave         bool
+	BrokerVersion int32
+}
+
+type (
+	// groupName of consumer
+	producerData string
+
+	consumeType string
+
+	ServiceState int
+)
+
+const (
+	StateCreateJust ServiceState = iota
+	StateStartFailed
+	StateRunning
+	StateShutdown
+)
+
+type SubscriptionData struct {
+	ClassFilterMode bool
+	Topic           string
+	SubString       string
+	Tags            map[string]bool
+	Codes           map[int32]bool
+	SubVersion      int64
+	ExpType         string
+}
+
+type consumerData struct {
+	GroupName         string              `json:"groupName"`
+	CType             consumeType         `json:"consumeType"`
+	MessageModel      string              `json:"messageModel"`
+	Where             string              `json:"consumeFromWhere"`
+	SubscriptionDatas []*SubscriptionData `json:"subscriptionDataSet"`
+	UnitMode          bool                `json:"unitMode"`
+}
+
+type heartbeatData struct {
+	ClientId      string         `json:"clientID"`
+	ProducerDatas []producerData `json:"producerDataSet"`
+	ConsumerDatas []consumerData `json:"consumerDataSet"`
+}
+
+func (data *heartbeatData) encode() []byte {
+	d, err := json.Marshal(data)
+	if err != nil {
+		rlog.Errorf("marshal heartbeatData error: %s", err.Error())
+		return nil
+	}
+	rlog.Info(string(d))
+	return d
+}
diff --git a/kernel/mq_version.go b/internal/kernel/mq_version.go
similarity index 100%
rename from kernel/mq_version.go
rename to internal/kernel/mq_version.go
diff --git a/kernel/perm.go b/internal/kernel/perm.go
similarity index 100%
rename from kernel/perm.go
rename to internal/kernel/perm.go
diff --git a/kernel/request.go b/internal/kernel/request.go
similarity index 100%
rename from kernel/request.go
rename to internal/kernel/request.go
diff --git a/kernel/response.go b/internal/kernel/response.go
similarity index 100%
rename from kernel/response.go
rename to internal/kernel/response.go
diff --git a/kernel/route.go b/internal/kernel/route.go
similarity index 96%
rename from kernel/route.go
rename to internal/kernel/route.go
index d0678e6..f216c0e 100644
--- a/kernel/route.go
+++ b/internal/kernel/route.go
@@ -20,7 +20,8 @@ package kernel
 import (
 	"encoding/json"
 	"errors"
-	"github.com/apache/rocketmq-client-go/remote"
+	"github.com/apache/rocketmq-client-go/internal/remote"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 	"github.com/apache/rocketmq-client-go/utils"
 	"github.com/tidwall/gjson"
@@ -98,7 +99,7 @@ func cleanOfflineBroker() {
 type TopicPublishInfo struct {
 	OrderTopic          bool
 	HaveTopicRouterInfo bool
-	MqList              []*MessageQueue
+	MqList              []*primitive.MessageQueue
 	RouteData           *TopicRouteData
 	TopicQueueIndex     int32
 }
@@ -221,19 +222,19 @@ func FindBrokerAddressInSubscribe(brokerName string, brokerId int64, onlyThisBro
 	return result
 }
 
-func FetchSubscribeMessageQueues(topic string) ([]*MessageQueue, error) {
+func FetchSubscribeMessageQueues(topic string) ([]*primitive.MessageQueue, error) {
 	routeData, err := queryTopicRouteInfoFromServer(topic)
 
 	if err != nil {
 		return nil, err
 	}
 
-	mqs := make([]*MessageQueue, 0)
+	mqs := make([]*primitive.MessageQueue, 0)
 
 	for _, qd := range routeData.QueueDataList {
 		if queueIsReadable(qd.Perm) {
 			for i := 0; i < qd.ReadQueueNums; i++ {
-				mqs = append(mqs, &MessageQueue{Topic: topic, BrokerName: qd.BrokerName, QueueId: i})
+				mqs = append(mqs, &primitive.MessageQueue{Topic: topic, BrokerName: qd.BrokerName, QueueId: i})
 			}
 		}
 	}
@@ -321,7 +322,7 @@ func routeData2PublishInfo(topic string, data *TopicRouteData) *TopicPublishInfo
 			item := strings.Split(broker, ":")
 			nums, _ := strconv.Atoi(item[1])
 			for i := 0; i < nums; i++ {
-				mq := &MessageQueue{
+				mq := &primitive.MessageQueue{
 					Topic:      topic,
 					BrokerName: item[0],
 					QueueId:    i,
@@ -357,7 +358,7 @@ func routeData2PublishInfo(topic string, data *TopicRouteData) *TopicPublishInfo
 		}
 
 		for i := 0; i < qd.WriteQueueNums; i++ {
-			mq := &MessageQueue{
+			mq := &primitive.MessageQueue{
 				Topic:      topic,
 				BrokerName: qd.BrokerName,
 				QueueId:    i,
diff --git a/kernel/route_test.go b/internal/kernel/route_test.go
similarity index 100%
rename from kernel/route_test.go
rename to internal/kernel/route_test.go
diff --git a/kernel/transaction.go b/internal/kernel/transaction.go
similarity index 100%
rename from kernel/transaction.go
rename to internal/kernel/transaction.go
diff --git a/kernel/validators.go b/internal/kernel/validators.go
similarity index 100%
rename from kernel/validators.go
rename to internal/kernel/validators.go
diff --git a/producer/producer.go b/internal/producer/producer.go
similarity index 85%
rename from producer/producer.go
rename to internal/producer/producer.go
index 8bb0bf8..52657ee 100644
--- a/producer/producer.go
+++ b/internal/producer/producer.go
@@ -21,8 +21,9 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/kernel"
-	"github.com/apache/rocketmq-client-go/remote"
+	"github.com/apache/rocketmq-client-go/internal/kernel"
+	"github.com/apache/rocketmq-client-go/internal/remote"
+	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
 	"github.com/apache/rocketmq-client-go/utils"
 	"os"
@@ -34,11 +35,11 @@ import (
 type Producer interface {
 	Start() error
 	Shutdown() error
-	SendSync(context.Context, *kernel.Message) (*kernel.SendResult, error)
-	SendOneWay(context.Context, *kernel.Message) error
+	SendSync(context.Context, *primitive.Message) (*primitive.SendResult, error)
+	SendOneWay(context.Context, *primitive.Message) error
 }
 
-func NewProducer(opt ProducerOptions) (Producer, error) {
+func NewProducer(opt primitive.ProducerOptions) (Producer, error) {
 	if err := utils.VerifyIP(opt.NameServerAddr); err != nil {
 		return nil, err
 	}
@@ -63,18 +64,10 @@ type defaultProducer struct {
 	group       string
 	client      *kernel.RMQClient
 	state       kernel.ServiceState
-	options     ProducerOptions
+	options     primitive.ProducerOptions
 	publishInfo sync.Map
 }
 
-type ProducerOptions struct {
-	kernel.ClientOption
-	NameServerAddr           string
-	GroupName                string
-	RetryTimesWhenSendFailed int
-	UnitMode                 bool
-}
-
 func (p *defaultProducer) Start() error {
 	p.state = kernel.StateRunning
 	p.client.RegisterProducer(p.group, p)
@@ -86,7 +79,7 @@ func (p *defaultProducer) Shutdown() error {
 	return nil
 }
 
-func (p *defaultProducer) SendSync(ctx context.Context, msg *kernel.Message) (*kernel.SendResult, error) {
+func (p *defaultProducer) SendSync(ctx context.Context, msg *primitive.Message) (*primitive.SendResult, error) {
 	if msg == nil {
 		return nil, errors.New("message is nil")
 	}
@@ -122,7 +115,7 @@ func (p *defaultProducer) SendSync(ctx context.Context, msg *kernel.Message) (*k
 	return nil, err
 }
 
-func (p *defaultProducer) SendOneWay(ctx context.Context, msg *kernel.Message) error {
+func (p *defaultProducer) SendOneWay(ctx context.Context, msg *primitive.Message) error {
 	if msg == nil {
 		return errors.New("message is nil")
 	}
@@ -157,7 +150,7 @@ func (p *defaultProducer) SendOneWay(ctx context.Context, msg *kernel.Message) e
 	return err
 }
 
-func (p *defaultProducer) buildSendRequest(mq *kernel.MessageQueue, msg *kernel.Message) *remote.RemotingCommand {
+func (p *defaultProducer) buildSendRequest(mq *primitive.MessageQueue, msg *primitive.Message) *remote.RemotingCommand {
 	req := &kernel.SendMessageRequest{
 		ProducerGroup:  p.group,
 		Topic:          mq.Topic,
@@ -173,7 +166,7 @@ func (p *defaultProducer) buildSendRequest(mq *kernel.MessageQueue, msg *kernel.
 	return remote.NewRemotingCommand(kernel.ReqSendMessage, req, msg.Body)
 }
 
-func (p *defaultProducer) selectMessageQueue(topic string) *kernel.MessageQueue {
+func (p *defaultProducer) selectMessageQueue(topic string) *primitive.MessageQueue {
 	v, exist := p.publishInfo.Load(topic)
 
 	if !exist {
diff --git a/remote/codec.go b/internal/remote/codec.go
similarity index 100%
rename from remote/codec.go
rename to internal/remote/codec.go
diff --git a/remote/codec_test.go b/internal/remote/codec_test.go
similarity index 100%
rename from remote/codec_test.go
rename to internal/remote/codec_test.go
diff --git a/remote/remote_client.go b/internal/remote/remote_client.go
similarity index 100%
rename from remote/remote_client.go
rename to internal/remote/remote_client.go
diff --git a/remote/remote_client_test.go b/internal/remote/remote_client_test.go
similarity index 100%
rename from remote/remote_client_test.go
rename to internal/remote/remote_client_test.go
diff --git a/remote/rpchook.go b/internal/remote/rpchook.go
similarity index 100%
rename from remote/rpchook.go
rename to internal/remote/rpchook.go
diff --git a/primitive/consume.go b/primitive/consume.go
new file mode 100644
index 0000000..40e89b9
--- /dev/null
+++ b/primitive/consume.go
@@ -0,0 +1,128 @@
+package primitive
+
+// Message model defines the way how messages are delivered to each consumer clients.
+// </p>
+//
+// RocketMQ supports two message models: clustering and broadcasting. If clustering is set, consumer clients with
+// the same {@link #consumerGroup} would only consume shards of the messages subscribed, which achieves load
+// balances; Conversely, if the broadcasting is set, each consumer client will consume all subscribed messages
+// separately.
+// </p>
+//
+// This field defaults to clustering.
+type MessageModel int
+
+const (
+	BroadCasting MessageModel = iota
+	Clustering
+)
+
+func (mode MessageModel) String() string {
+	switch mode {
+	case BroadCasting:
+		return "BroadCasting"
+	case Clustering:
+		return "Clustering"
+	default:
+		return "Unknown"
+	}
+}
+
+// Consuming point on consumer booting.
+// </p>
+//
+// There are three consuming points:
+// <ul>
+// <li>
+// <code>CONSUME_FROM_LAST_OFFSET</code>: consumer clients pick up where it stopped previously.
+// If it were a newly booting up consumer client, according aging of the consumer group, there are two
+// cases:
+// <ol>
+// <li>
+// if the consumer group is created so recently that the earliest message being subscribed has yet
+// expired, which means the consumer group represents a lately launched business, consuming will
+// start from the very beginning;
+// </li>
+// <li>
+// if the earliest message being subscribed has expired, consuming will start from the latest
+// messages, meaning messages born prior to the booting timestamp would be ignored.
+// </li>
+// </ol>
+// </li>
+// <li>
+// <code>CONSUME_FROM_FIRST_OFFSET</code>: Consumer client will start from earliest messages available.
+// </li>
+// <li>
+// <code>CONSUME_FROM_TIMESTAMP</code>: Consumer client will start from specified timestamp, which means
+// messages born prior to {@link #consumeTimestamp} will be ignored
+// </li>
+// </ul>
+type ConsumeFromWhere int
+
+const (
+	ConsumeFromLastOffset ConsumeFromWhere = iota
+	ConsumeFromFirstOffset
+	ConsumeFromTimestamp
+)
+
+type ExpressionType string
+
+const (
+	/**
+	 * <ul>
+	 * Keywords:
+	 * <li>{@code AND, OR, NOT, BETWEEN, IN, TRUE, FALSE, IS, NULL}</li>
+	 * </ul>
+	 * <p/>
+	 * <ul>
+	 * Data type:
+	 * <li>Boolean, like: TRUE, FALSE</li>
+	 * <li>String, like: 'abc'</li>
+	 * <li>Decimal, like: 123</li>
+	 * <li>Float number, like: 3.1415</li>
+	 * </ul>
+	 * <p/>
+	 * <ul>
+	 * Grammar:
+	 * <li>{@code AND, OR}</li>
+	 * <li>{@code >, >=, <, <=, =}</li>
+	 * <li>{@code BETWEEN A AND B}, equals to {@code >=A AND <=B}</li>
+	 * <li>{@code NOT BETWEEN A AND B}, equals to {@code >B OR <A}</li>
+	 * <li>{@code IN ('a', 'b')}, equals to {@code ='a' OR ='b'}, this operation only support String type.</li>
+	 * <li>{@code IS NULL}, {@code IS NOT NULL}, check parameter whether is null, or not.</li>
+	 * <li>{@code =TRUE}, {@code =FALSE}, check parameter whether is true, or false.</li>
+	 * </ul>
+	 * <p/>
+	 * <p>
+	 * Example:
+	 * (a > 10 AND a < 100) OR (b IS NOT NULL AND b=TRUE)
+	 * </p>
+	 */
+	SQL92 = ExpressionType("SQL92")
+
+	/**
+	 * Only support or operation such as
+	 * "tag1 || tag2 || tag3", <br>
+	 * If null or * expression, meaning subscribe all.
+	 */
+	TAG = ExpressionType("TAG")
+)
+
+func IsTagType(exp string) bool {
+	if exp == "" || exp == "TAG" {
+		return true
+	}
+	return false
+}
+
+type MessageSelector struct {
+	Type       ExpressionType
+	Expression string
+}
+
+type ConsumeResult int
+
+const (
+	ConsumeSuccess ConsumeResult = iota
+	ConsumeRetryLater
+)
diff --git a/kernel/message.go b/primitive/message.go
similarity index 85%
rename from kernel/message.go
rename to primitive/message.go
index 0f65853..d15c8a3 100644
--- a/kernel/message.go
+++ b/primitive/message.go
@@ -15,9 +15,12 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kernel
+package primitive
 
-import "fmt"
+import (
+	"fmt"
+	"github.com/apache/rocketmq-client-go/utils"
+)
 
 const (
 	PropertyKeySeparator                   = " "
@@ -118,3 +121,28 @@ func (msgExt *MessageExt) String() string {
 		msgExt.StoreTimestamp, msgExt.StoreHost, msgExt.CommitLogOffset, msgExt.BodyCRC, msgExt.ReconsumeTimes,
 		msgExt.PreparedTransactionOffset)
 }
+
+// MessageQueue message queue
+type MessageQueue struct {
+	Topic      string `json:"topic"`
+	BrokerName string `json:"brokerName"`
+	QueueId    int    `json:"queueId"`
+}
+
+func (mq *MessageQueue) String() string {
+	return fmt.Sprintf("MessageQueue [topic=%s, brokerName=%s, queueId=%d]", mq.Topic, mq.BrokerName, mq.QueueId)
+}
+
+func (mq *MessageQueue) HashCode() int {
+	result := 1
+	result = 31*result + utils.HashString(mq.BrokerName)
+	result = 31*result + mq.QueueId
+	result = 31*result + utils.HashString(mq.Topic)
+
+	return result
+}
+
+func (mq *MessageQueue) Equals(queue *MessageQueue) bool {
+	// TODO
+	return true
+}
diff --git a/primitive/options.go b/primitive/options.go
new file mode 100644
index 0000000..b657ac0
--- /dev/null
+++ b/primitive/options.go
@@ -0,0 +1,132 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package primitive
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"time"
+)
+
+type ProducerOptions struct {
+	ClientOption
+	NameServerAddr           string
+	GroupName                string
+	RetryTimesWhenSendFailed int
+	UnitMode                 bool
+}
+
+type ConsumerOption struct {
+	ClientOption
+	NameServerAddr string
+
+	/**
+	 * Backtracking consumption time with second precision. Time format is
+	 * 20131223171201<br>
+	 * Implying Seventeen twelve and 01 seconds on December 23, 2013 year<br>
+	 * Default backtracking consumption time Half an hour ago.
+	 */
+	ConsumeTimestamp string
+
+	// The socket timeout in milliseconds
+	ConsumerPullTimeout time.Duration
+
+	// Concurrently max span offset.it has no effect on sequential consumption
+	ConsumeConcurrentlyMaxSpan int
+
+	// Flow control threshold on queue level, each message queue will cache at most 1000 messages by default,
+	// Consider the {PullBatchSize}, the instantaneous value may exceed the limit
+	PullThresholdForQueue int64
+
+	// Limit the cached message size on queue level, each message queue will cache at most 100 MiB messages by default,
+	// Consider the {@code pullBatchSize}, the instantaneous value may exceed the limit
+	//
+	// The size of a message only measured by message body, so it's not accurate
+	PullThresholdSizeForQueue int
+
+	// Flow control threshold on topic level, default value is -1(Unlimited)
+	//
+	// The value of {@code pullThresholdForQueue} will be overwrote and calculated based on
+	// {@code pullThresholdForTopic} if it is't unlimited
+	//
+	// For example, if the value of pullThresholdForTopic is 1000 and 10 message queues are assigned to this consumer,
+	// then pullThresholdForQueue will be set to 100
+	PullThresholdForTopic int
+
+	// Limit the cached message size on topic level, default value is -1 MiB(Unlimited)
+	//
+	// The value of {@code pullThresholdSizeForQueue} will be overwrote and calculated based on
+	// {@code pullThresholdSizeForTopic} if it is't unlimited
+	//
+	// For example, if the value of pullThresholdSizeForTopic is 1000 MiB and 10 message queues are
+	// assigned to this consumer, then pullThresholdSizeForQueue will be set to 100 MiB
+	PullThresholdSizeForTopic int
+
+	// Message pull Interval
+	PullInterval time.Duration
+
+	// Batch consumption size
+	ConsumeMessageBatchMaxSize int
+
+	// Batch pull size
+	PullBatchSize int32
+
+	// Whether update subscription relationship when every pull
+	PostSubscriptionWhenPull bool
+
+	// Max re-consume times. -1 means 16 times.
+	//
+	// If messages are re-consumed more than {@link #maxReconsumeTimes} before success, it's be directed to a deletion
+	// queue waiting.
+	MaxReconsumeTimes int
+
+	// Suspending pulling time for cases requiring slow pulling like flow-control scenario.
+	SuspendCurrentQueueTimeMillis time.Duration
+
+	// Maximum amount of time a message may block the consuming thread.
+	ConsumeTimeout time.Duration
+
+	ConsumerModel  MessageModel
+	Strategy       AllocateStrategy
+	ConsumeOrderly bool
+	FromWhere      ConsumeFromWhere
+	// TODO traceDispatcher
+}
+
+func (opt *ClientOption) ChangeInstanceNameToPID() {
+	if opt.InstanceName == "DEFAULT" {
+		opt.InstanceName = strconv.Itoa(os.Getegid())
+	}
+}
+
+func (opt *ClientOption) String() string {
+	return fmt.Sprintf("ClientOption [ClientIP=%s, InstanceName=%s, "+
+		"UnitMode=%v, UnitName=%s, VIPChannelEnabled=%v, UseTLS=%v]", opt.ClientIP,
+		opt.InstanceName, opt.UnitMode, opt.UnitName, opt.VIPChannelEnabled, opt.UseTLS)
+}
+
+type ClientOption struct {
+	NameServerAddr    string
+	ClientIP          string
+	InstanceName      string
+	UnitMode          bool
+	UnitName          string
+	VIPChannelEnabled bool
+	UseTLS            bool
+}
diff --git a/kernel/model.go b/primitive/result.go
similarity index 71%
rename from kernel/model.go
rename to primitive/result.go
index 8ed0f97..0c1e738 100644
--- a/kernel/model.go
+++ b/primitive/result.go
@@ -15,14 +15,12 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package kernel
+package primitive
 
 import (
 	"bytes"
 	"encoding/binary"
-	"encoding/json"
 	"fmt"
-	"github.com/apache/rocketmq-client-go/rlog"
 	"github.com/apache/rocketmq-client-go/utils"
 )
 
@@ -209,85 +207,3 @@ func toMessages(messageExts []*MessageExt) []*Message {
 
 	return msgs
 }
-
-// MessageQueue message queue
-type MessageQueue struct {
-	Topic      string `json:"topic"`
-	BrokerName string `json:"brokerName"`
-	QueueId    int    `json:"queueId"`
-}
-
-func (mq *MessageQueue) String() string {
-	return fmt.Sprintf("MessageQueue [topic=%s, brokerName=%s, queueId=%d]", mq.Topic, mq.BrokerName, mq.QueueId)
-}
-
-func (mq *MessageQueue) HashCode() int {
-	result := 1
-	result = 31*result + utils.HashString(mq.BrokerName)
-	result = 31*result + mq.QueueId
-	result = 31*result + utils.HashString(mq.Topic)
-
-	return result
-}
-
-func (mq *MessageQueue) Equals(queue *MessageQueue) bool {
-	// TODO
-	return true
-}
-
-type FindBrokerResult struct {
-	BrokerAddr    string
-	Slave         bool
-	BrokerVersion int32
-}
-
-type (
-	// groupName of consumer
-	producerData string
-
-	consumeType string
-
-	ServiceState int
-)
-
-const (
-	StateCreateJust ServiceState = iota
-	StateStartFailed
-	StateRunning
-	StateShutdown
-)
-
-type SubscriptionData struct {
-	ClassFilterMode bool
-	Topic           string
-	SubString       string
-	Tags            map[string]bool
-	Codes           map[int32]bool
-	SubVersion      int64
-	ExpType         string
-}
-
-type consumerData struct {
-	GroupName         string              `json:"groupName"`
-	CType             consumeType         `json:"consumeType"`
-	MessageModel      string              `json:"messageModel"`
-	Where             string              `json:"consumeFromWhere"`
-	SubscriptionDatas []*SubscriptionData `json:"subscriptionDataSet"`
-	UnitMode          bool                `json:"unitMode"`
-}
-
-type heartbeatData struct {
-	ClientId      string         `json:"clientID"`
-	ProducerDatas []producerData `json:"producerDataSet"`
-	ConsumerDatas []consumerData `json:"consumerDataSet"`
-}
-
-func (data *heartbeatData) encode() []byte {
-	d, err := json.Marshal(data)
-	if err != nil {
-		rlog.Errorf("marshal heartbeatData error: %s", err.Error())
-		return nil
-	}
-	rlog.Info(string(d))
-	return d
-}
diff --git a/primitive/strategy.go b/primitive/strategy.go
new file mode 100644
index 0000000..56dfef1
--- /dev/null
+++ b/primitive/strategy.go
@@ -0,0 +1,117 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one or more
+contributor license agreements.  See the NOTICE file distributed with
+this work for additional information regarding copyright ownership.
+The ASF licenses this file to You under the Apache License, Version 2.0
+(the "License"); you may not use this file except in compliance with
+the License.  You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package primitive
+
+import (
+	"github.com/apache/rocketmq-client-go/rlog"
+	"github.com/apache/rocketmq-client-go/utils"
+)
+
+// Strategy Algorithm for message allocating between consumers
+// An allocate strategy proxy for based on machine room nearside priority. An actual allocate strategy can be
+// specified.
+//
+// If any consumer is alive in a machine room, the message queue of the broker which is deployed in the same machine
+// should only be allocated to those. Otherwise, those message queues can be shared along all consumers since there are
+// no alive consumer to monopolize them.
+//
+// Average Hashing queue algorithm
+// Cycle average Hashing queue algorithm
+// Use Message Queue specified
+// Computer room Hashing queue algorithm, such as Alipay logic room
+// Consistent Hashing queue algorithm
+
+type AllocateStrategy func(string, string, []*MessageQueue, []string) []*MessageQueue
+
+func AllocateByAveragely(consumerGroup, currentCID string, mqAll []*MessageQueue,
+	cidAll []string) []*MessageQueue {
+	if currentCID == "" || utils.IsArrayEmpty(mqAll) || utils.IsArrayEmpty(cidAll) {
+		return nil
+	}
+	var (
+		find  bool
+		index int
+	)
+
+	for idx := range cidAll {
+		if cidAll[idx] == currentCID {
+			find = true
+			index = idx
+			break
+		}
+	}
+	if !find {
+		rlog.Infof("[BUG] ConsumerGroup=%s, ConsumerId=%s not in cidAll:%+v", consumerGroup, currentCID, cidAll)
+		return nil
+	}
+
+	mqSize := len(mqAll)
+	cidSize := len(cidAll)
+	mod := mqSize % cidSize
+
+	var averageSize int
+	if mqSize <= cidSize {
+		averageSize = 1
+	} else {
+		if mod > 0 && index < mod {
+			averageSize = mqSize/cidSize + 1
+		} else {
+			averageSize = mqSize / cidSize
+		}
+	}
+
+	var startIndex int
+	if mod > 0 && index < mod {
+		startIndex = index * averageSize
+	} else {
+		startIndex = index*averageSize + mod
+	}
+
+	num := utils.MinInt(averageSize, mqSize-startIndex)
+	result := make([]*MessageQueue, num)
+	for i := 0; i < num; i++ {
+		result[i] = mqAll[(startIndex+i)%mqSize]
+	}
+	return result
+}
+
+// TODO
+func AllocateByMachineNearby(consumerGroup, currentCID string, mqAll []*MessageQueue,
+	cidAll []string) []*MessageQueue {
+	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+}
+
+func AllocateByAveragelyCircle(consumerGroup, currentCID string, mqAll []*MessageQueue,
+	cidAll []string) []*MessageQueue {
+	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+}
+
+func AllocateByConfig(consumerGroup, currentCID string, mqAll []*MessageQueue,
+	cidAll []string) []*MessageQueue {
+	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+}
+
+func AllocateByMachineRoom(consumerGroup, currentCID string, mqAll []*MessageQueue,
+	cidAll []string) []*MessageQueue {
+	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+}
+
+func AllocateByConsistentHash(consumerGroup, currentCID string, mqAll []*MessageQueue,
+	cidAll []string) []*MessageQueue {
+	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+}