This is an automated email from the ASF dual-hosted git repository.
mytang0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 2c3d73eee [ISSUE #3290]Refactor ConsumerManager
new 86aa68824 Merge pull request #3291 from mxsm/eventmesh-3290
2c3d73eee is described below
commit 2c3d73eeec666c4154d99f17766bd9750db0e40a
Author: mxsm <[email protected]>
AuthorDate: Mon Feb 27 23:52:10 2023 +0800
[ISSUE #3290]Refactor ConsumerManager
---
.../protocol/http/consumer/ConsumerManager.java | 207 +++++----------------
1 file changed, 48 insertions(+), 159 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerManager.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerManager.java
index 0f24a1cfa..ffc74a722 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerManager.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/consumer/ConsumerManager.java
@@ -26,8 +26,6 @@ import org.apache.eventmesh.runtime.util.EventMeshUtil;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
import com.google.common.eventbus.Subscribe;
@@ -37,20 +35,15 @@ import lombok.extern.slf4j.Slf4j;
@Slf4j
public class ConsumerManager {
- private EventMeshHTTPServer eventMeshHTTPServer;
+ private final EventMeshHTTPServer eventMeshHTTPServer;
/**
* consumerGroup to ConsumerGroupManager.
*/
- private ConcurrentHashMap<String, ConsumerGroupManager> consumerTable =
- new ConcurrentHashMap<>();
+ private ConcurrentHashMap<String, ConsumerGroupManager> consumerTable =
new ConcurrentHashMap<>(64);
- private static final int DEFAULT_UPDATE_TIME = 3 * 30 * 1000;
- private ScheduledExecutorService scheduledExecutorService =
- Executors.newSingleThreadScheduledExecutor();
-
- public ConsumerManager(EventMeshHTTPServer eventMeshHTTPServer) {
+ public ConsumerManager(final EventMeshHTTPServer eventMeshHTTPServer) {
this.eventMeshHTTPServer = eventMeshHTTPServer;
}
@@ -61,114 +54,15 @@ public class ConsumerManager {
public void start() throws Exception {
log.info("consumerManager started......");
-
- // scheduledExecutorService.scheduleAtFixedRate(() -> {
- // logger.info("clientInfo check start.....");
- // synchronized
(eventMeshHTTPServer.localClientInfoMapping) {
- // Map<String, List<Client>> clientInfoMap =
- // eventMeshHTTPServer.localClientInfoMapping;
- // if (clientInfoMap.size() > 0) {
- // for (String key : clientInfoMap.keySet()) {
- // String consumerGroup = key.split("@")[0];
- // String topic = key.split("@")[1];
- // List<Client> clientList =
clientInfoMap.get(key);
- // Iterator<Client> clientIterator =
clientList.iterator();
- // boolean isChange = false;
- // while (clientIterator.hasNext()) {
- // Client client = clientIterator.next();
- // //The time difference is greater than 3
heartbeat cycles
- // if (System.currentTimeMillis() -
client.lastUpTime.getTime()
- // > DEFAULT_UPDATE_TIME) {
- // logger.warn(
- // "client {} lastUpdate time {}
over three heartbeat cycles",
- // JsonUtils.serialize(client),
client.lastUpTime);
- // clientIterator.remove();
- // isChange = true;
- // }
- // }
- // if (isChange) {
- // if (clientList.size() > 0) {
- // //change url
- // logger.info("consumerGroup {} client
info changing",
- // consumerGroup);
- // Map<String, List<String>> idcUrls =
new HashMap<>();
- // Set<String> clientUrls = new
HashSet<>();
- // for (Client client : clientList) {
- // clientUrls.add(client.url);
- // if
(idcUrls.containsKey(client.idc)) {
- // idcUrls.get(client.idc)
- //
.add(StringUtils.deleteWhitespace(client.url));
- // } else {
- // List<String> urls = new
ArrayList<>();
- // urls.add(client.url);
- // idcUrls.put(client.idc,
urls);
- // }
- // }
- // synchronized
(eventMeshHTTPServer.localConsumerGroupMapping) {
- // ConsumerGroupConf
consumerGroupConf =
- //
eventMeshHTTPServer.localConsumerGroupMapping
- // .get(consumerGroup);
- // Map<String,
ConsumerGroupTopicConf> map =
- //
consumerGroupConf.getConsumerGroupTopicConf();
- // for (String topicKey :
map.keySet()) {
- // if
(StringUtils.equals(topic, topicKey)) {
- // ConsumerGroupTopicConf
latestTopicConf =
- // new
ConsumerGroupTopicConf();
- //
latestTopicConf.setConsumerGroup(consumerGroup);
- //
latestTopicConf.setTopic(topic);
- //
latestTopicConf.setSubscriptionItem(
- //
map.get(topicKey).getSubscriptionItem());
- //
latestTopicConf.setUrls(clientUrls);
- //
- //
latestTopicConf.setIdcUrls(idcUrls);
- //
- // map.put(topic,
latestTopicConf);
- // }
- // }
- //
eventMeshHTTPServer.localConsumerGroupMapping
- // .put(consumerGroup,
consumerGroupConf);
- // logger.info(
- // "consumerGroup {} client
info changed, "
- // + "consumerGroupConf
{}", consumerGroup,
- //
JsonUtils.serialize(consumerGroupConf));
- //
- // try {
- //
notifyConsumerManager(consumerGroup, consumerGroupConf);
- // } catch (Exception e) {
- //
logger.error("notifyConsumerManager error", e);
- // }
- // }
- //
- // } else {
- // logger.info("consumerGroup {} client
info removed",
- // consumerGroup);
- // //remove
- // try {
- //
notifyConsumerManager(consumerGroup, null);
- // } catch (Exception e) {
- //
logger.error("notifyConsumerManager error", e);
- // }
- //
- //
eventMeshHTTPServer.localConsumerGroupMapping.keySet()
- // .removeIf(s ->
StringUtils.equals(consumerGroup, s));
- // }
- // }
- //
- // }
- // }
- // }
- // }, 10000, 10000, TimeUnit.MILLISECONDS);
- //TODO: update the subscription periodically from registry
}
/**
* notify ConsumerManager groupLevel
*/
- public void notifyConsumerManager(String consumerGroup,
- ConsumerGroupConf latestConsumerGroupConfig)
- throws Exception {
- ConsumerGroupManager cgm =
-
eventMeshHTTPServer.getConsumerManager().getConsumer(consumerGroup);
+ public void notifyConsumerManager(String consumerGroup, ConsumerGroupConf
latestConsumerGroupConfig) throws Exception {
+
+ ConsumerGroupManager cgm =
eventMeshHTTPServer.getConsumerManager().getConsumer(consumerGroup);
+
if (latestConsumerGroupConfig == null) {
ConsumerGroupStateEvent notification = new
ConsumerGroupStateEvent();
notification.action =
ConsumerGroupStateEvent.ConsumerGroupStateAction.DELETE;
@@ -219,10 +113,8 @@ public class ConsumerManager {
* @param consumerGroupConfig
* @throws Exception
*/
- public synchronized void addConsumer(String consumerGroup,
- ConsumerGroupConf consumerGroupConfig) throws Exception {
- ConsumerGroupManager cgm =
- new ConsumerGroupManager(eventMeshHTTPServer, consumerGroupConfig);
+ public synchronized void addConsumer(String consumerGroup,
ConsumerGroupConf consumerGroupConfig) throws Exception {
+ ConsumerGroupManager cgm = new
ConsumerGroupManager(eventMeshHTTPServer, consumerGroupConfig);
cgm.init();
cgm.start();
consumerTable.put(consumerGroup, cgm);
@@ -256,8 +148,7 @@ public class ConsumerManager {
log.info("start delConsumer with consumerGroup {}", consumerGroup);
if (consumerTable.containsKey(consumerGroup)) {
ConsumerGroupManager cgm = consumerTable.remove(consumerGroup);
- log.info("start unsubscribe topic with consumer group manager {}",
- JsonUtils.toJSONString(cgm));
+ log.info("start unsubscribe topic with consumer group manager {}",
JsonUtils.toJSONString(cgm));
cgm.unsubscribe(consumerGroup);
cgm.shutdown();
}
@@ -265,38 +156,36 @@ public class ConsumerManager {
}
@Subscribe
- public void onChange(ConsumerGroupTopicConfChangeEvent event) {
+ public void
handleConsumerGroupTopicConfChangeEvent(ConsumerGroupTopicConfChangeEvent
event) {
try {
log.info("onChange event:{}", event);
- if (event.action
- ==
ConsumerGroupTopicConfChangeEvent.ConsumerGroupTopicConfChangeAction.NEW) {
- ConsumerGroupManager manager =
getConsumer(event.consumerGroup);
- if (Objects.isNull(manager)) {
- return;
+ switch (event.action) {
+ case NEW: {
+ ConsumerGroupManager manager =
getConsumer(event.consumerGroup);
+ if (Objects.isNull(manager)) {
+ return;
+ }
+
manager.getConsumerGroupConfig().getConsumerGroupTopicConf().put(event.topic,
event.newTopicConf);
+ break;
}
- manager.getConsumerGroupConfig().getConsumerGroupTopicConf()
- .put(event.topic, event.newTopicConf);
- return;
- }
-
- if (event.action
- ==
ConsumerGroupTopicConfChangeEvent.ConsumerGroupTopicConfChangeAction.CHANGE) {
- ConsumerGroupManager manager =
getConsumer(event.consumerGroup);
- if (Objects.isNull(manager)) {
- return;
+ case CHANGE: {
+ ConsumerGroupManager manager =
getConsumer(event.consumerGroup);
+ if (Objects.isNull(manager)) {
+ return;
+ }
+
manager.getConsumerGroupConfig().getConsumerGroupTopicConf().replace(event.topic,
event.newTopicConf);
+ break;
}
- manager.getConsumerGroupConfig().getConsumerGroupTopicConf()
- .replace(event.topic, event.newTopicConf);
- return;
- }
-
- if (event.action
- ==
ConsumerGroupTopicConfChangeEvent.ConsumerGroupTopicConfChangeAction.DELETE) {
- ConsumerGroupManager manager =
getConsumer(event.consumerGroup);
- if (Objects.isNull(manager)) {
- return;
+ case DELETE: {
+ ConsumerGroupManager manager =
getConsumer(event.consumerGroup);
+ if (Objects.isNull(manager)) {
+ return;
+ }
+
manager.getConsumerGroupConfig().getConsumerGroupTopicConf().remove(event.topic);
+ break;
}
-
manager.getConsumerGroupConfig().getConsumerGroupTopicConf().remove(event.topic);
+ default:
+ //do nothing
}
} catch (Exception ex) {
log.error("onChange event:{} err", event, ex);
@@ -304,22 +193,22 @@ public class ConsumerManager {
}
@Subscribe
- public void onChange(ConsumerGroupStateEvent event) {
+ public void handleConsumerGroupStateEvent(ConsumerGroupStateEvent event) {
try {
log.info("onChange event:{}", event);
- if (event.action ==
ConsumerGroupStateEvent.ConsumerGroupStateAction.NEW) {
- addConsumer(event.consumerGroup, event.consumerGroupConfig);
- return;
- }
-
- if (event.action ==
ConsumerGroupStateEvent.ConsumerGroupStateAction.CHANGE) {
- restartConsumer(event.consumerGroup,
event.consumerGroupConfig);
- return;
- }
- if (event.action ==
ConsumerGroupStateEvent.ConsumerGroupStateAction.DELETE) {
- delConsumer(event.consumerGroup);
- return;
+ switch (event.action) {
+ case NEW:
+ addConsumer(event.consumerGroup,
event.consumerGroupConfig);
+ break;
+ case CHANGE:
+ restartConsumer(event.consumerGroup,
event.consumerGroupConfig);
+ break;
+ case DELETE:
+ delConsumer(event.consumerGroup);
+ break;
+ default:
+ //do nothing
}
} catch (Exception ex) {
log.error("onChange event:{} err", event, ex);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
