Repository: apache/rocketmq-client-go
Pull request #141 (2019-08-03 16:08:38)

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



Patch:
diff --git a/.travis.yml b/.travis.yml
index 82f760c..ea08a34 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -13,6 +13,7 @@ env:
     - TOPIC=test
     - GROUP=testGroup
     - GO111MODULE=on
+    - HOME=$TRAVIS_HOME
   matrix:
     - OS_TYPE=centos OS_VERSION=7
 
@@ -34,4 +35,4 @@ script:
   - go mod vendor && go test ./... -coverprofile=coverage.txt -covermode=atomic
 
 after_success:
-  - bash <(curl -s https://codecov.io/bash)
\ No newline at end of file
+  - bash <(curl -s https://codecov.io/bash)
diff --git a/consumer/option.go b/consumer/option.go
index dafff61..558fd90 100644
--- a/consumer/option.go
+++ b/consumer/option.go
@@ -202,3 +202,9 @@ func WithMaxReconsumeTimes(times int32) Option {
 		opts.MaxReconsumeTimes = times
 	}
 }
+
+func WithStrategy(strategy AllocateStrategy) Option {
+	return func(opts *consumerOptions) {
+		opts.Strategy = strategy
+	}
+}
diff --git a/consumer/strategy.go b/consumer/strategy.go
index c5b6a25..6920149 100644
--- a/consumer/strategy.go
+++ b/consumer/strategy.go
@@ -21,6 +21,7 @@ import (
 	"github.com/apache/rocketmq-client-go/internal/utils"
 	"github.com/apache/rocketmq-client-go/primitive"
 	"github.com/apache/rocketmq-client-go/rlog"
+	"strings"
 )
 
 // Strategy Algorithm for message allocating between consumers
@@ -41,14 +42,14 @@ type AllocateStrategy func(string, string, []*primitive.MessageQueue, []string)
 
 func AllocateByAveragely(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
 	cidAll []string) []*primitive.MessageQueue {
-	if currentCID == "" || utils.IsArrayEmpty(mqAll) || utils.IsArrayEmpty(cidAll) {
+	if currentCID == "" || len(mqAll) == 0 || len(cidAll) == 0 {
 		return nil
 	}
+
 	var (
 		find  bool
 		index int
 	)
-
 	for idx := range cidAll {
 		if cidAll[idx] == currentCID {
 			find = true
@@ -57,7 +58,7 @@ func AllocateByAveragely(consumerGroup, currentCID string, mqAll []*primitive.Me
 		}
 	}
 	if !find {
-		rlog.Infof("[BUG] ConsumerGroup=%s, ConsumerId=%s not in cidAll:%+v", consumerGroup, currentCID, cidAll)
+		rlog.Warnf("[BUG] ConsumerGroup=%s, ConsumerId=%s not in cidAll:%+v", consumerGroup, currentCID, cidAll)
 		return nil
 	}
 
@@ -84,32 +85,104 @@ func AllocateByAveragely(consumerGroup, currentCID string, mqAll []*primitive.Me
 	}
 
 	num := utils.MinInt(averageSize, mqSize-startIndex)
-	result := make([]*primitive.MessageQueue, num)
+	result := []*primitive.MessageQueue{}
 	for i := 0; i < num; i++ {
-		result[i] = mqAll[(startIndex+i)%mqSize]
+		result = append(result, mqAll[(startIndex+i)%mqSize])
 	}
 	return result
 }
 
-// TODO
-func AllocateByMachineNearby(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
+func AllocateByAveragelyCircle(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
 	cidAll []string) []*primitive.MessageQueue {
-	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+	if currentCID == "" || len(mqAll) == 0 || len(cidAll) == 0 {
+		return nil
+	}
+
+	var (
+		find  bool
+		index int
+	)
+	for idx := range cidAll {
+		if cidAll[idx] == currentCID {
+			find = true
+			index = idx
+			break
+		}
+	}
+	if !find {
+		rlog.Warnf("[BUG] ConsumerGroup=%s, ConsumerId=%s not in cidAll:%+v", consumerGroup, currentCID, cidAll)
+		return nil
+	}
+
+	result := []*primitive.MessageQueue{}
+	for i := index; i < len(mqAll); i++ {
+		if i%len(cidAll) == index {
+			result = append(result, mqAll[i])
+		}
+	}
+	return result
 }
 
-func AllocateByAveragelyCircle(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
+// TODO
+func AllocateByMachineNearby(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
 	cidAll []string) []*primitive.MessageQueue {
 	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
 }
 
-func AllocateByConfig(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
-	cidAll []string) []*primitive.MessageQueue {
-	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+func AllocateByConfig(list []*primitive.MessageQueue) AllocateStrategy {
+	return func(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, cidAll []string) []*primitive.MessageQueue {
+		return list
+	}
 }
 
-func AllocateByMachineRoom(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
-	cidAll []string) []*primitive.MessageQueue {
-	return AllocateByAveragely(consumerGroup, currentCID, mqAll, cidAll)
+func AllocateByMachineRoom(consumeridcs []string) AllocateStrategy {
+	return func(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue, cidAll []string) []*primitive.MessageQueue {
+		if currentCID == "" || len(mqAll) == 0 || len(cidAll) == 0 {
+			return nil
+		}
+
+		var (
+			find  bool
+			index int
+		)
+		for idx := range cidAll {
+			if cidAll[idx] == currentCID {
+				find = true
+				index = idx
+				break
+			}
+		}
+		if !find {
+			rlog.Warnf("[BUG] ConsumerGroup=%s, ConsumerId=%s not in cidAll:%+v", consumerGroup, currentCID, cidAll)
+			return nil
+		}
+
+		premqAll := []*primitive.MessageQueue{}
+		for _, mq := range mqAll {
+			temp := strings.Split(mq.BrokerName, "@")
+			if len(temp) == 2 {
+				for _, idc := range consumeridcs {
+					if idc == temp[0] {
+						premqAll = append(premqAll, mq)
+					}
+				}
+			}
+		}
+
+		mod := len(premqAll) / len(cidAll)
+		rem := len(premqAll) % len(cidAll)
+		startIndex := mod * index
+		endIndex := startIndex + mod
+
+		result := []*primitive.MessageQueue{}
+		for i := startIndex; i < endIndex; i++ {
+			result = append(result, mqAll[i])
+		}
+		if rem > index {
+			result = append(result, premqAll[index+mod*len(cidAll)])
+		}
+		return result
+	}
 }
 
 func AllocateByConsistentHash(consumerGroup, currentCID string, mqAll []*primitive.MessageQueue,
diff --git a/docs/feature.md b/docs/feature.md
index 9caa789..795ffac 100644
--- a/docs/feature.md
+++ b/docs/feature.md
@@ -40,9 +40,9 @@
     
 ### AllocateMessageQueueStrategy
 - [x] AllocateMessageQueueAveragely
-- [ ] AllocateMessageQueueAveragelyByCircle
-- [ ] AllocateMessageQueueByConfig
-- [ ] AllocateMessageQueueByMachineRoom
+- [x] AllocateMessageQueueAveragelyByCircle
+- [X] AllocateMessageQueueByConfig
+- [X] AllocateMessageQueueByMachineRoom
 
 ### Other
 - [x] Rebalance
diff --git a/examples/consumer/strategy/main.go b/examples/consumer/strategy/main.go
new file mode 100644
index 0000000..4483c0d
--- /dev/null
+++ b/examples/consumer/strategy/main.go
@@ -0,0 +1,52 @@
+/*
+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 main
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/apache/rocketmq-client-go"
+	"github.com/apache/rocketmq-client-go/consumer"
+	"github.com/apache/rocketmq-client-go/primitive"
+)
+
+func main() {
+	c, _ := rocketmq.NewPushConsumer(
+		consumer.WithGroupName("testGroup"),
+		consumer.WithNameServer([]string{"127.0.0.1:9876"}),
+		consumer.WithStrategy(consumer.AllocateByAveragely),
+	)
+	err := c.Subscribe("TopicTest", consumer.MessageSelector{}, func(ctx context.Context,
+		msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
+		fmt.Printf("subscribe callback: %v \n", msgs)
+		return consumer.ConsumeSuccess, nil
+	})
+	if err != nil {
+		fmt.Println(err.Error())
+	}
+	// Note: start after subscribe
+	err = c.Start()
+	if err != nil {
+		fmt.Println(err.Error())
+		os.Exit(-1)
+	}
+	time.Sleep(time.Hour)
+}
diff --git a/go.mod b/go.mod
index f271d91..a15098e 100644
--- a/go.mod
+++ b/go.mod
@@ -1,6 +1,7 @@
 module github.com/apache/rocketmq-client-go
 
 require (
+	github.com/agiledragon/gomonkey v0.0.0-20190517145658-8fa491f7b918
 	github.com/emirpasic/gods v1.12.0
 	github.com/golang/mock v1.3.1
 	github.com/pkg/errors v0.8.1
diff --git a/go.sum b/go.sum
index cc4c4b6..82cd61d 100644
--- a/go.sum
+++ b/go.sum
@@ -1,3 +1,5 @@
+github.com/agiledragon/gomonkey v0.0.0-20190517145658-8fa491f7b918 h1:a88Ln+jbIokfi6xoKtq10dbgp4VMg1CmHF1J42p8EyE=
+github.com/agiledragon/gomonkey v0.0.0-20190517145658-8fa491f7b918/go.mod h1:2NGfXu1a80LLr2cmWXGBDaHEjb1idR6+FVlX5T3D9hw=
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
@@ -41,4 +43,5 @@ golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/tools v0.0.0-20190328211700-ab21143f2384/go.mod h1:LCzVGOaR6xXOjkQ3onu1FJEFr0SW1gC7cKk1uF8kGRs=
+golang.org/x/tools v0.0.0-20190425150028-36563e24a262 h1:qsl9y/CJx34tuA7QCPNp86JNJe4spst6Ff8MjvPUdPg=
 golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
diff --git a/internal/utils/helper.go b/internal/utils/helper.go
index 2b9eb2a..58ea791 100644
--- a/internal/utils/helper.go
+++ b/internal/utils/helper.go
@@ -24,7 +24,3 @@ func GetAddressByBytes(data []byte) string {
 func UnCompress(data []byte) []byte {
 	return data
 }
-
-func IsArrayEmpty(i ...interface{}) bool {
-	return len(i) == 0
-}


Test patch:
diff --git a/consumer/offset_store_test.go b/consumer/offset_store_test.go
new file mode 100644
index 0000000..6bfc90c
--- /dev/null
+++ b/consumer/offset_store_test.go
@@ -0,0 +1,282 @@
+package consumer
+
+import (
+	"github.com/agiledragon/gomonkey"
+	"github.com/apache/rocketmq-client-go/internal"
+	"github.com/apache/rocketmq-client-go/internal/remote"
+	"github.com/apache/rocketmq-client-go/primitive"
+	"github.com/golang/mock/gomock"
+	. "github.com/smartystreets/goconvey/convey"
+	"testing"
+)
+
+func TestNewLocalFileOffsetStore(t *testing.T) {
+	Convey("Given some test cases", t, func() {
+		type testCase struct {
+			clientId       string
+			group          string
+			expectedResult *localFileOffsetStore
+		}
+		cases := []testCase{
+			{
+				clientId: "",
+				group:    "testGroup",
+				expectedResult: &localFileOffsetStore{
+					group: "testGroup",
+					path:  _LocalOffsetStorePath + "/testGroup/offset.json",
+				},
+			}, {
+				clientId: "192.168.24.1@default",
+				group:    "",
+				expectedResult: &localFileOffsetStore{
+					group: "",
+					path:  _LocalOffsetStorePath + "/192.168.24.1@default/offset.json",
+				},
+			}, {
+				clientId: "192.168.24.1@default",
+				group:    "testGroup",
+				expectedResult: &localFileOffsetStore{
+					group: "testGroup",
+					path:  _LocalOffsetStorePath + "/192.168.24.1@default/testGroup/offset.json",
+				},
+			},
+		}
+
+		for _, value := range cases {
+			result := NewLocalFileOffsetStore(value.clientId, value.group).(*localFileOffsetStore)
+			value.expectedResult.OffsetTable = result.OffsetTable
+			So(result, ShouldResemble, value.expectedResult)
+		}
+	})
+}
+
+func TestReadFromMemory(t *testing.T) {
+	Convey("Given some queue and queueOffset with a starting value", t, func() {
+		type testCase struct {
+			queueOffset    map[string]map[int]*queueOffset
+			queue          *primitive.MessageQueue
+			expectedOffset int64
+		}
+
+		cases := []testCase{
+			{
+				queueOffset: map[string]map[int]*queueOffset{
+					"testTopic2": {
+						1: {
+							QueueID: 1,
+							Broker:  "default",
+							Offset:  1,
+						},
+					},
+				},
+				queue: &primitive.MessageQueue{
+					Topic:      "testTopic1",
+					BrokerName: "default",
+					QueueId:    1,
+				},
+				expectedOffset: -1,
+			},
+			{
+				queueOffset: map[string]map[int]*queueOffset{
+					"testTopic1": {
+						0: {
+							QueueID: 0,
+							Broker:  "default",
+							Offset:  1,
+						},
+					},
+				},
+				queue: &primitive.MessageQueue{
+					Topic:      "testTopic1",
+					BrokerName: "default",
+					QueueId:    1,
+				},
+				expectedOffset: -1,
+			},
+			{
+				queueOffset: map[string]map[int]*queueOffset{
+					"testTopic1": {
+						0: {
+							QueueID: 0,
+							Broker:  "default",
+							Offset:  1,
+						},
+					},
+				},
+				queue: &primitive.MessageQueue{
+					Topic:      "testTopic1",
+					BrokerName: "default",
+					QueueId:    0,
+				},
+				expectedOffset: 1,
+			},
+		}
+
+		for _, value := range cases {
+			offset := readFromMemory(value.queueOffset, value.queue)
+			So(offset, ShouldEqual, value.expectedOffset)
+		}
+	})
+}
+
+func TestLocalFileOffsetStore(t *testing.T) {
+	Convey("Given a local store with a starting value", t, func() {
+		localStore := NewLocalFileOffsetStore("192.168.24.1@default", "testGroup")
+
+		type offsetCase struct {
+			queue          *primitive.MessageQueue
+			setOffset      int64
+			expectedOffset int64
+		}
+		mq := &primitive.MessageQueue{
+			Topic:      "testTopic",
+			BrokerName: "default",
+			QueueId:    1,
+		}
+
+		Convey("test update", func() {
+			Convey("when increaseOnly is false", func() {
+				cases := []offsetCase{
+					{
+						queue:          mq,
+						setOffset:      3,
+						expectedOffset: 3,
+					}, {
+						queue:          mq,
+						setOffset:      1,
+						expectedOffset: 1,
+					},
+				}
+				for _, value := range cases {
+					localStore.update(value.queue, value.setOffset, false)
+					offset := localStore.read(value.queue, _ReadFromMemory)
+					So(offset, ShouldEqual, value.expectedOffset)
+				}
+			})
+
+			Convey("when increaseOnly is true", func() {
+				localStore.update(mq, 0, false)
+
+				cases := []offsetCase{
+					{
+						queue:          mq,
+						setOffset:      3,
+						expectedOffset: 3,
+					}, {
+						queue:          mq,
+						setOffset:      1,
+						expectedOffset: 3,
+					},
+				}
+				for _, value := range cases {
+					localStore.update(value.queue, value.setOffset, true)
+					offset := localStore.read(value.queue, _ReadFromMemory)
+					So(offset, ShouldEqual, value.expectedOffset)
+				}
+			})
+		})
+
+		Convey("test persist", func() {
+			localStore.update(mq, 1, false)
+
+			queues := []*primitive.MessageQueue{mq}
+			localStore.persist(queues)
+			offset := localStore.read(mq, _ReadFromStore)
+			So(offset, ShouldEqual, 1)
+		})
+	})
+}
+
+func TestRemoteBrokerOffsetStore(t *testing.T) {
+	Convey("Given a remote store with a starting value", t, func() {
+		ctrl := gomock.NewController(t)
+		defer ctrl.Finish()
+
+		rmqClient := internal.NewMockRMQClient(ctrl)
+		remoteStore := NewRemoteOffsetStore("testGroup", rmqClient)
+
+		type offsetCase struct {
+			queue          *primitive.MessageQueue
+			setOffset      int64
+			expectedOffset int64
+		}
+		mq := &primitive.MessageQueue{
+			Topic:      "testTopic",
+			BrokerName: "default",
+			QueueId:    1,
+		}
+
+		Convey("test update", func() {
+			Convey("when increaseOnly is false", func() {
+				cases := []offsetCase{
+					{
+						queue:          mq,
+						setOffset:      3,
+						expectedOffset: 3,
+					}, {
+						queue:          mq,
+						setOffset:      1,
+						expectedOffset: 1,
+					},
+				}
+				for _, value := range cases {
+					remoteStore.update(value.queue, value.setOffset, false)
+					offset := remoteStore.read(value.queue, _ReadFromMemory)
+					So(offset, ShouldEqual, value.expectedOffset)
+				}
+			})
+
+			Convey("when increaseOnly is true", func() {
+				remoteStore.update(mq, 0, false)
+
+				cases := []offsetCase{
+					{
+						queue:          mq,
+						setOffset:      3,
+						expectedOffset: 3,
+					}, {
+						queue:          mq,
+						setOffset:      1,
+						expectedOffset: 3,
+					},
+				}
+				for _, value := range cases {
+					remoteStore.update(value.queue, value.setOffset, true)
+					offset := remoteStore.read(value.queue, _ReadFromMemory)
+					So(offset, ShouldEqual, value.expectedOffset)
+				}
+			})
+		})
+
+		Convey("test persist", func() {
+			queues := []*primitive.MessageQueue{mq}
+
+			patch := gomonkey.ApplyFunc(internal.FindBrokerAddrByName, func(_ string) string {
+				return "192.168.24.1:10911"
+			})
+			defer patch.Reset()
+
+			ret := &remote.RemotingCommand{
+				Code: internal.ResSuccess,
+				ExtFields: map[string]string{
+					"offset": "1",
+				},
+			}
+			rmqClient.EXPECT().InvokeSync(gomock.Any(), gomock.Any(), gomock.Any()).Return(ret, nil)
+
+			remoteStore.persist(queues)
+			offset := remoteStore.read(mq, _ReadFromStore)
+			So(offset, ShouldEqual, 1)
+		})
+
+		Convey("test remove", func() {
+			remoteStore.update(mq, 1, false)
+			offset := remoteStore.read(mq, _ReadFromMemory)
+			So(offset, ShouldEqual, 1)
+
+			remoteStore.remove(mq)
+			offset = remoteStore.read(mq, _ReadFromMemory)
+			So(offset, ShouldEqual, -1)
+		})
+	})
+}
diff --git a/consumer/strategy_test.go b/consumer/strategy_test.go
new file mode 100644
index 0000000..53214ce
--- /dev/null
+++ b/consumer/strategy_test.go
@@ -0,0 +1,379 @@
+package consumer
+
+import (
+	"github.com/apache/rocketmq-client-go/primitive"
+	. "github.com/smartystreets/goconvey/convey"
+	"testing"
+)
+
+func TestAllocateByAveragely(t *testing.T) {
+	Convey("Given message queues with a starting value", t, func() {
+		queues := []*primitive.MessageQueue{
+			{
+				QueueId: 0,
+			},
+			{
+				QueueId: 1,
+			},
+			{
+				QueueId: 2,
+			},
+			{
+				QueueId: 3,
+			},
+			{
+				QueueId: 4,
+			},
+			{
+				QueueId: 5,
+			},
+		}
+
+		Convey("When params is empty", func() {
+			result := AllocateByAveragely("testGroup", "", queues, []string{"192.168.24.1@default"})
+			So(result, ShouldBeNil)
+
+			result = AllocateByAveragely("testGroup", "192.168.24.1@default", nil, []string{"192.168.24.1@default"})
+			So(result, ShouldBeNil)
+
+			result = AllocateByAveragely("testGroup", "192.168.24.1@default", queues, nil)
+			So(result, ShouldBeNil)
+		})
+
+		type testCase struct {
+			currentCid    string
+			mqAll         []*primitive.MessageQueue
+			cidAll        []string
+			expectedQueue []*primitive.MessageQueue
+		}
+		cases := []testCase{
+			{
+				currentCid: "192.168.24.1@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 0,
+					},
+					{
+						QueueId: 1,
+					},
+					{
+						QueueId: 2,
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.2@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 2,
+					},
+					{
+						QueueId: 3,
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.2@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 2,
+					},
+					{
+						QueueId: 3,
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.4@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 5,
+					},
+				},
+			},
+			{
+				currentCid:    "192.168.24.7@default",
+				mqAll:         queues,
+				cidAll:        []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default", "192.168.24.5@default", "192.168.24.6@default", "192.168.24.7@default"},
+				expectedQueue: []*primitive.MessageQueue{},
+			},
+		}
+
+		Convey("the result of AllocateByAveragely should be deep equal expectedQueue", func() {
+			for _, value := range cases {
+				result := AllocateByAveragely("testGroup", value.currentCid, value.mqAll, value.cidAll)
+				So(result, ShouldResemble, value.expectedQueue)
+			}
+		})
+	})
+}
+
+func TestAllocateByAveragelyCircle(t *testing.T) {
+	Convey("Given message queues with a starting value", t, func() {
+		queues := []*primitive.MessageQueue{
+			{
+				QueueId: 0,
+			},
+			{
+				QueueId: 1,
+			},
+			{
+				QueueId: 2,
+			},
+			{
+				QueueId: 3,
+			},
+			{
+				QueueId: 4,
+			},
+			{
+				QueueId: 5,
+			},
+		}
+
+		Convey("When params is empty", func() {
+			result := AllocateByAveragelyCircle("testGroup", "", queues, []string{"192.168.24.1@default"})
+			So(result, ShouldBeNil)
+
+			result = AllocateByAveragelyCircle("testGroup", "192.168.24.1@default", nil, []string{"192.168.24.1@default"})
+			So(result, ShouldBeNil)
+
+			result = AllocateByAveragelyCircle("testGroup", "192.168.24.1@default", queues, nil)
+			So(result, ShouldBeNil)
+		})
+
+		type testCase struct {
+			currentCid    string
+			mqAll         []*primitive.MessageQueue
+			cidAll        []string
+			expectedQueue []*primitive.MessageQueue
+		}
+		cases := []testCase{
+			{
+				currentCid: "192.168.24.1@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 0,
+					},
+					{
+						QueueId: 2,
+					},
+					{
+						QueueId: 4,
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.2@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 1,
+					},
+					{
+						QueueId: 4,
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.2@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 1,
+					},
+					{
+						QueueId: 5,
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.4@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId: 3,
+					},
+				},
+			},
+			{
+				currentCid:    "192.168.24.7@default",
+				mqAll:         queues,
+				cidAll:        []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default", "192.168.24.5@default", "192.168.24.6@default", "192.168.24.7@default"},
+				expectedQueue: []*primitive.MessageQueue{},
+			},
+		}
+
+		Convey("the result of AllocateByAveragelyCircle should be deep equal expectedQueue", func() {
+			for _, value := range cases {
+				result := AllocateByAveragelyCircle("testGroup", value.currentCid, value.mqAll, value.cidAll)
+				So(result, ShouldResemble, value.expectedQueue)
+			}
+		})
+	})
+}
+
+func TestAllocateByConfig(t *testing.T) {
+	Convey("Given message queues with a starting value", t, func() {
+		queues := []*primitive.MessageQueue{
+			{
+				QueueId: 0,
+			},
+			{
+				QueueId: 1,
+			},
+			{
+				QueueId: 2,
+			},
+			{
+				QueueId: 3,
+			},
+			{
+				QueueId: 4,
+			},
+			{
+				QueueId: 5,
+			},
+		}
+
+		strategy := AllocateByConfig(queues)
+		result := strategy("testGroup", "192.168.24.1@default", queues, []string{"192.168.24.1@default", "192.168.24.2@default"})
+		So(result, ShouldResemble, queues)
+	})
+}
+
+func TestAllocateByMachineRoom(t *testing.T) {
+	Convey("Given some consumer IDCs with a starting value", t, func() {
+		idcs := []string{"192.168.24.1", "192.168.24.2"}
+		strategy := AllocateByMachineRoom(idcs)
+
+		queues := []*primitive.MessageQueue{
+			{
+				QueueId:    0,
+				BrokerName: "192.168.24.1@defaultName",
+			},
+			{
+				QueueId:    1,
+				BrokerName: "192.168.24.1@defaultName",
+			},
+			{
+				QueueId:    2,
+				BrokerName: "192.168.24.1@defaultName",
+			},
+			{
+				QueueId:    3,
+				BrokerName: "192.168.24.2@defaultName",
+			},
+			{
+				QueueId:    4,
+				BrokerName: "192.168.24.2@defaultName",
+			},
+			{
+				QueueId:    5,
+				BrokerName: "192.168.24.3@defaultName",
+			},
+		}
+
+		Convey("When params is empty", func() {
+			result := strategy("testGroup", "", queues, []string{"192.168.24.1@default"})
+			So(result, ShouldBeNil)
+
+			result = strategy("testGroup", "192.168.24.1@default", nil, []string{"192.168.24.1@default"})
+			So(result, ShouldBeNil)
+
+			result = strategy("testGroup", "192.168.24.1@default", queues, nil)
+			So(result, ShouldBeNil)
+		})
+
+		type testCase struct {
+			currentCid    string
+			mqAll         []*primitive.MessageQueue
+			cidAll        []string
+			expectedQueue []*primitive.MessageQueue
+		}
+		cases := []testCase{
+			{
+				currentCid: "192.168.24.1@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId:    0,
+						BrokerName: "192.168.24.1@defaultName",
+					},
+					{
+						QueueId:    1,
+						BrokerName: "192.168.24.1@defaultName",
+					},
+					{
+						QueueId:    4,
+						BrokerName: "192.168.24.2@defaultName",
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.2@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId:    1,
+						BrokerName: "192.168.24.1@defaultName",
+					},
+					{
+						QueueId:    4,
+						BrokerName: "192.168.24.2@defaultName",
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.2@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId:    1,
+						BrokerName: "192.168.24.1@defaultName",
+					},
+				},
+			},
+			{
+				currentCid: "192.168.24.4@default",
+				mqAll:      queues,
+				cidAll:     []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default"},
+				expectedQueue: []*primitive.MessageQueue{
+					{
+						QueueId:    3,
+						BrokerName: "192.168.24.2@defaultName",
+					},
+				},
+			},
+			{
+				currentCid:    "192.168.24.7@default",
+				mqAll:         queues,
+				cidAll:        []string{"192.168.24.1@default", "192.168.24.2@default", "192.168.24.3@default", "192.168.24.4@default", "192.168.24.5@default", "192.168.24.6@default", "192.168.24.7@default"},
+				expectedQueue: []*primitive.MessageQueue{},
+			},
+		}
+
+		Convey("the result of AllocateByMachineRoom should be deep equal expectedQueue", func() {
+			for _, value := range cases {
+				result := strategy("testGroup", value.currentCid, value.mqAll, value.cidAll)
+				So(result, ShouldResemble, value.expectedQueue)
+			}
+		})
+	})
+}