提交 5c5180ec 编写于 作者: D dongeforever

Remove diamond operator for client module with JDK 1.6

上级 1f4b893c
......@@ -76,19 +76,19 @@ public class AllocateMessageQueueConsistentHash implements AllocateMessageQueue
}
Collection<ClientNode> cidNodes = new ArrayList<>();
Collection<ClientNode> cidNodes = new ArrayList<ClientNode>();
for (String cid : cidAll) {
cidNodes.add(new ClientNode(cid));
}
final ConsistentHashRouter<ClientNode> router; //for building hash ring
if (customHashFunction != null) {
router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt, customHashFunction);
router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt, customHashFunction);
} else {
router = new ConsistentHashRouter<>(cidNodes, virtualNodeCnt);
router = new ConsistentHashRouter<ClientNode>(cidNodes, virtualNodeCnt);
}
List<MessageQueue> results = new ArrayList<>();
List<MessageQueue> results = new ArrayList<MessageQueue>();
for (MessageQueue mq : mqAll) {
ClientNode clientNode = router.routeNode(mq.toString());
if (clientNode != null && currentCID.equals(clientNode.getKey())) {
......
......@@ -23,6 +23,7 @@ import java.util.Map;
import java.util.Random;
import java.util.TreeMap;
import org.apache.rocketmq.client.consumer.AllocateMessageQueueStrategy;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Assert;
import org.junit.Before;
......@@ -113,13 +114,13 @@ public class AllocateMessageQueueConsitentHashTest {
//System.out.println("mqAll:" + mqAll.toString());
List<String> cidAll = createConsumerIdList(consumerSize);
List<MessageQueue> allocatedResAll = new ArrayList<>();
List<MessageQueue> allocatedResAll = new ArrayList<MessageQueue>();
Map<MessageQueue, String> allocateToAllOrigin = new TreeMap<>();
Map<MessageQueue, String> allocateToAllOrigin = new TreeMap<MessageQueue, String>();
//test allocate all
{
List<String> cidBegin = new ArrayList<>(cidAll);
List<String> cidBegin = new ArrayList<String>(cidAll);
//System.out.println("cidAll:" + cidBegin.toString());
for (String cid : cidBegin) {
......@@ -135,13 +136,13 @@ public class AllocateMessageQueueConsitentHashTest {
verifyAllocateAll(cidBegin,mqAll, allocatedResAll));
}
Map<MessageQueue, String> allocateToAllAfterRemoveOne = new TreeMap<>();
List<String> cidAfterRemoveOne = new ArrayList<>(cidAll);
Map<MessageQueue, String> allocateToAllAfterRemoveOne = new TreeMap<MessageQueue, String>();
List<String> cidAfterRemoveOne = new ArrayList<String>(cidAll);
//test allocate remove one cid
{
String removeCID = cidAfterRemoveOne.remove(0);
//System.out.println("removing one cid "+removeCID);
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<>();
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<MessageQueue>();
Iterator<Map.Entry<MessageQueue, String>> it = allocateToAllOrigin.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<MessageQueue, String> entry = it.next();
......@@ -151,7 +152,7 @@ public class AllocateMessageQueueConsitentHashTest {
}
//System.out.println("cidAll:" + cidAfterRemoveOne.toString());
List<MessageQueue> allocatedResAllAfterRemove = new ArrayList<>();
List<MessageQueue> allocatedResAllAfterRemove = new ArrayList<MessageQueue>();
for (String cid : cidAfterRemoveOne) {
List<MessageQueue> rs = allocateMessageQueueConsistentHash.allocate("testConsumerGroup", cid, mqAll, cidAfterRemoveOne);
allocatedResAllAfterRemove.addAll(rs);
......@@ -166,16 +167,16 @@ public class AllocateMessageQueueConsitentHashTest {
verifyAfterRemove(allocateToAllOrigin, allocateToAllAfterRemoveOne, removeCID);
}
List<String> cidAfterAdd = new ArrayList<>(cidAfterRemoveOne);
List<String> cidAfterAdd = new ArrayList<String>(cidAfterRemoveOne);
//test allocate add one more cid
{
String newCid = CID_PREFIX+"NEW";
//System.out.println("add one more cid "+newCid);
cidAfterAdd.add(newCid);
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<>();
List<MessageQueue> mqShouldOnlyChanged = new ArrayList<MessageQueue>();
//System.out.println("cidAll:" + cidAfterAdd.toString());
List<MessageQueue> allocatedResAllAfterAdd = new ArrayList<>();
Map<MessageQueue, String> allocateToAll3 = new TreeMap<>();
List<MessageQueue> allocatedResAllAfterAdd = new ArrayList<MessageQueue>();
Map<MessageQueue, String> allocateToAll3 = new TreeMap<MessageQueue, String>();
for (String cid : cidAfterAdd) {
List<MessageQueue> rs = allocateMessageQueueConsistentHash.allocate("testConsumerGroup", cid, mqAll, cidAfterAdd);
allocatedResAllAfterAdd.addAll(rs);
......@@ -225,7 +226,7 @@ public class AllocateMessageQueueConsitentHashTest {
}
private List<String> createConsumerIdList(int size) {
List<String> consumerIdList = new ArrayList<>(size);
List<String> consumerIdList = new ArrayList<String>(size);
for (int i = 0; i < size; i++) {
consumerIdList.add(CID_PREFIX + String.valueOf(i));
}
......@@ -233,7 +234,7 @@ public class AllocateMessageQueueConsitentHashTest {
}
private List<MessageQueue> createMessageQueueList(int size) {
List<MessageQueue> messageQueueList = new ArrayList<>(size);
List<MessageQueue> messageQueueList = new ArrayList<MessageQueue>(size);
for (int i = 0; i < size; i++) {
MessageQueue mq = new MessageQueue(topic, "brokerName", i);
messageQueueList.add(mq);
......
......@@ -30,7 +30,7 @@ import java.util.TreeMap;
* @param <T>
*/
public class ConsistentHashRouter<T extends Node> {
private final SortedMap<Long, VirtualNode<T>> ring = new TreeMap<>();
private final SortedMap<Long, VirtualNode<T>> ring = new TreeMap<Long, VirtualNode<T>>();
private final HashFunction hashFunction;
public ConsistentHashRouter(Collection<T> pNodes, int vNodeCount) {
......@@ -64,7 +64,7 @@ public class ConsistentHashRouter<T extends Node> {
if (vNodeCount < 0) throw new IllegalArgumentException("illegal virtual node counts :" + vNodeCount);
int existingReplicas = getExistingReplicas(pNode);
for (int i = 0; i < vNodeCount; i++) {
VirtualNode<T> vNode = new VirtualNode<>(pNode, i + existingReplicas);
VirtualNode<T> vNode = new VirtualNode<T>(pNode, i + existingReplicas);
ring.put(hashFunction.hash(vNode.getKey()), vNode);
}
}
......
......@@ -722,7 +722,7 @@ public class CommitLog {
messageExtBatch.setEncodedBuff(batchEncoder.encode(messageExtBatch));
lockForPutMessage(); //spin...
putMessageLock.lock();
try {
long beginLockTimestamp = this.defaultMessageStore.getSystemClock().now();
this.beginTimeInLock = beginLockTimestamp;
......@@ -771,7 +771,7 @@ public class CommitLog {
eclipseTimeInLock = this.defaultMessageStore.getSystemClock().now() - beginLockTimestamp;
beginTimeInLock = 0;
} finally {
releasePutMessageLock();
putMessageLock.unlock();
}
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册