opensource/rocketmq
3
0
Fork 0
mirror of https://github.com/apache/rocketmq.git synced 2026-09-21 13:49:50 +08:00
Code Issues Packages Projects Releases Wiki Activity

[ISSUE #3949] refactor SelectableMessageQueue and do the merging work.

Browse Source
This commit is contained in:
Jixiang.jjx
2022-07-13 11:29:09 +08:00
committed by zhouxiang
parent 0e5dee8b4f
commit be512d5747
4 changed files with 6 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
proxy/src/main/java/org/apache/rocketmq/proxy/client/TopicRouteCache.java
+1 -1
View File
@@ -71,7 +71,7 @@ public class TopicRouteCache {
if (last == null) {
return getMessageQueue(topic).getWrite().selectOne(false);
}
return getMessageQueue(topic).getWrite().selectNextOne(last);
return getMessageQueue(topic).getWrite().selectNextQueue(last);
}
public AddressableMessageQueue selectOneWriteQueue(String topic, String brokerName, int queueId) throws Exception {
proxy/src/main/java/org/apache/rocketmq/proxy/client/route/SelectableMessageQueue.java
+3 -21
View File
@@ -119,7 +119,7 @@ public class SelectableMessageQueue {
AddressableMessageQueue mq = new AddressableMessageQueue(
new MessageQueue(topicRoute.getTopicName(), qd.getBrokerName(), i),
brokerAddr);
queueSet.add(mq);
queueSet.add(mq);
}
}
}
@@ -128,8 +128,6 @@ public class SelectableMessageQueue {
return queueSet.stream().sorted().collect(Collectors.toList());
}
// 这里应该不是线程安全的
private void buildBrokerActingQueues(String topic, List<AddressableMessageQueue> normalQueues) {
for (AddressableMessageQueue mq : normalQueues) {
AddressableMessageQueue brokerActingQueue = new AddressableMessageQueue(
@@ -194,28 +192,12 @@ public class SelectableMessageQueue {
return newOne;
}
// should use selectNextQueue
// public final AddressableMessageQueue selectNextBrokerActingQueue(AddressableMessageQueue last) {
// boolean onlyBroker = last.getQueueId() < 0;
// AddressableMessageQueue newOne = last;
// int count = onlyBroker ? brokerActingQueues.size() : queues.size();
//
// for (int i = 0; i < count; i++) {
// newOne = selectOne(onlyBroker);
// if (!newOne.getBrokerName().equals(last.getBrokerName())) {
// break;
// }
// }
//
// return newOne;
// }
public List<AddressableMessageQueue> getQueues() {
return queues;
}
public List<AddressableMessageQueue> getBrokers() {
return brokers;
public List<AddressableMessageQueue> getBrokerActingQueues() {
return brokerActingQueues;
}
@Override
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/ConsumerService.java
+1 -1
View File
@@ -29,6 +29,6 @@ public class ConsumerService extends BaseService {
}
public CompletableFuture<ReceiveMessageResponse> receiveMessage(Context ctx, ReceiveMessageRequest request) {
return null;
}
}
proxy/src/main/java/org/apache/rocketmq/proxy/grpc/service/cluster/RouteService.java
+1 -1
View File
@@ -70,7 +70,7 @@ public class RouteService extends BaseService {
public List<AddressableMessageQueue> getAssignment(QueryAssignmentRequest request) throws Exception {
MessageQueueWrapper messageQueueWrapper = clientManager.getTopicRouteCache()
.getMessageQueue(Converter.getResourceNameWithNamespace(request.getTopic()));
return messageQueueWrapper.getRead().getBrokers();
return messageQueueWrapper.getRead().getBrokerActingQueues();
}
}