You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by yu...@apache.org on 2017/08/01 07:26:50 UTC
[25/50] [abbrv] incubator-rocketmq git commit: Remove diamond
operator for client module with JDK 1.6
Remove diamond operator for client module with JDK 1.6
Project: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/commit/8d781757
Tree: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/tree/8d781757
Diff: http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/diff/8d781757
Branch: refs/heads/master
Commit: 8d781757dfb56ebd95b0494e8e2c87422c22d767
Parents: e57f9ac
Author: dongeforever <zh...@yeah.net>
Authored: Sat May 27 14:09:08 2017 +0800
Committer: dongeforever <zh...@yeah.net>
Committed: Sat May 27 14:09:08 2017 +0800
----------------------------------------------------------------------
.../AllocateMessageQueueConsistentHash.java | 8 +++---
.../AllocateMessageQueueConsitentHashTest.java | 27 ++++++++++----------
.../consistenthash/ConsistentHashRouter.java | 4 +--
.../org/apache/rocketmq/store/CommitLog.java | 4 +--
4 files changed, 22 insertions(+), 21 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/8d781757/client/src/main/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsistentHash.java
----------------------------------------------------------------------
diff --git a/client/src/main/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsistentHash.java b/client/src/main/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsistentHash.java
index 77198b7..09d940a 100644
--- a/client/src/main/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsistentHash.java
+++ b/client/src/main/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsistentHash.java
@@ -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())) {
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/8d781757/client/src/test/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java
----------------------------------------------------------------------
diff --git a/client/src/test/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java b/client/src/test/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java
index fc7ab9f..e9e5db7 100644
--- a/client/src/test/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java
+++ b/client/src/test/java/org/apache/rocketmq/client/consumer/rebalance/AllocateMessageQueueConsitentHashTest.java
@@ -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);
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/8d781757/common/src/main/java/org/apache/rocketmq/common/consistenthash/ConsistentHashRouter.java
----------------------------------------------------------------------
diff --git a/common/src/main/java/org/apache/rocketmq/common/consistenthash/ConsistentHashRouter.java b/common/src/main/java/org/apache/rocketmq/common/consistenthash/ConsistentHashRouter.java
index 8606c43..a6fce51 100644
--- a/common/src/main/java/org/apache/rocketmq/common/consistenthash/ConsistentHashRouter.java
+++ b/common/src/main/java/org/apache/rocketmq/common/consistenthash/ConsistentHashRouter.java
@@ -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);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-rocketmq/blob/8d781757/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
----------------------------------------------------------------------
diff --git a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
index 7b29263..b44211c 100644
--- a/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
+++ b/store/src/main/java/org/apache/rocketmq/store/CommitLog.java
@@ -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();
}