This is an automated email from the ASF dual-hosted git repository.
mikexue pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new 08fabd20e [ISSUE #4060]Use AbstractRetryer to unify GrpcRetryer,
HttpRetryer and EventMeshTcpRetryer
08fabd20e is described below
commit 08fabd20ec6fcf23ad8197eb21b9f9ffec96e3ea
Author: pandaapo <[email protected]>
AuthorDate: Fri Sep 22 17:48:48 2023 +0800
[ISSUE #4060]Use AbstractRetryer to unify GrpcRetryer, HttpRetryer and
EventMeshTcpRetryer
---
.../runtime/core/protocol/AbstractRetryer.java | 87 ++++++++++++++++++++++
.../core/protocol/grpc/retry/GrpcRetryer.java | 57 ++------------
.../core/protocol/http/retry/HttpRetryer.java | 63 ++--------------
.../client/session/retry/EventMeshTcpRetryer.java | 51 ++-----------
.../runtime/metrics/grpc/EventMeshGrpcMonitor.java | 2 +-
.../runtime/metrics/http/HTTPMetricsServer.java | 4 +-
6 files changed, 112 insertions(+), 152 deletions(-)
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/AbstractRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/AbstractRetryer.java
new file mode 100644
index 000000000..3bfa13f87
--- /dev/null
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/AbstractRetryer.java
@@ -0,0 +1,87 @@
+/*
+ * 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 org.apache.eventmesh.runtime.core.protocol;
+
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import lombok.extern.slf4j.Slf4j;
+
+@Slf4j
+public abstract class AbstractRetryer {
+
+ protected final DelayQueue<DelayRetryable> retrys = new DelayQueue<>();
+
+ protected ThreadPoolExecutor pool;
+
+ protected Thread dispatcher;
+
+ protected abstract void pushRetry(DelayRetryable delayRetryable);
+
+ protected abstract void init();
+
+ public int getRetrySize() {
+ return retrys.size();
+ }
+
+ protected void initDispatcher(Thread dispatcher) {
+ dispatcher = new Thread(() -> {
+ try {
+ DelayRetryable retryObj;
+ while (!Thread.currentThread().isInterrupted() && (retryObj =
retrys.take()) != null) {
+ final DelayRetryable delayRetryable = retryObj;
+ pool.execute(() -> {
+ try {
+ delayRetryable.retry();
+ if (log.isDebugEnabled()) {
+ log.debug("retryObj : {}", delayRetryable);
+ }
+ } catch (Exception e) {
+ log.error("retry-dispatcher error!", e);
+ }
+ });
+ }
+ } catch (Exception e) {
+ if (e instanceof InterruptedException) {
+ Thread.currentThread().interrupt();
+ }
+ log.error("retry-dispatcher error!", e);
+ }
+ }, "retry-dispatcher");
+ dispatcher.setDaemon(true);
+ log.info("EventMesh retryer inited......");
+ }
+
+ public void shutdown() {
+ dispatcher.interrupt();
+ pool.shutdown();
+ log.info("EventMesh retryer shutdown......");
+ }
+
+ public void start() throws Exception {
+ dispatcher.start();
+ log.info("EventMesh retryer started......");
+ }
+
+ /**
+ * Get fail-retry queue, this method is just used for metrics.
+ */
+ public DelayQueue<DelayRetryable> getRetryQueue() {
+ return retrys;
+ }
+}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
index 6de4eee58..b3f209448 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/grpc/retry/GrpcRetryer.java
@@ -20,17 +20,17 @@ package
org.apache.eventmesh.runtime.core.protocol.grpc.retry;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.runtime.boot.EventMeshGrpcServer;
import org.apache.eventmesh.runtime.configuration.EventMeshGrpcConfiguration;
+import org.apache.eventmesh.runtime.core.protocol.AbstractRetryer;
import org.apache.eventmesh.runtime.core.protocol.DelayRetryable;
import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class GrpcRetryer {
+public class GrpcRetryer extends AbstractRetryer {
private final EventMeshGrpcConfiguration grpcConfiguration;
@@ -38,20 +38,16 @@ public class GrpcRetryer {
this.grpcConfiguration =
eventMeshGrpcServer.getEventMeshGrpcConfiguration();
}
- private final DelayQueue<DelayRetryable> failed = new
DelayQueue<DelayRetryable>();
-
- private ThreadPoolExecutor pool;
-
- private Thread dispatcher;
-
+ @Override
public void pushRetry(DelayRetryable delayRetryable) {
- if (failed.size() >=
grpcConfiguration.getEventMeshServerRetryBlockQueueSize()) {
+ if (retrys.size() >=
grpcConfiguration.getEventMeshServerRetryBlockQueueSize()) {
log.error("[RETRY-QUEUE] is full!");
return;
}
- failed.offer(delayRetryable);
+ retrys.offer(delayRetryable);
}
+ @Override
public void init() {
pool = new ThreadPoolExecutor(
grpcConfiguration.getEventMeshServerRetryThreadNum(),
@@ -60,46 +56,7 @@ public class GrpcRetryer {
new EventMeshThreadFactory("grpc-retry", true,
Thread.NORM_PRIORITY),
new ThreadPoolExecutor.AbortPolicy());
- dispatcher = new Thread(() -> {
- try {
- DelayRetryable retryObj;
- while (!Thread.currentThread().isInterrupted()
- && (retryObj = failed.take()) != null) {
- final DelayRetryable delayRetryable = retryObj;
- pool.execute(() -> {
- try {
- delayRetryable.retry();
- if (log.isDebugEnabled()) {
- log.debug("retryObj : {}", delayRetryable);
- }
- } catch (Exception e) {
- log.error("grpc-retry-dispatcher error!", e);
- }
- });
- }
- } catch (Exception e) {
- if (e instanceof InterruptedException) {
- Thread.currentThread().interrupt();
- }
- log.error("grpc-retry-dispatcher error!", e);
- }
- }, "grpc-retry-dispatcher");
- dispatcher.setDaemon(true);
- log.info("GrpcRetryer inited......");
- }
-
- public int size() {
- return failed.size();
- }
-
- public void shutdown() {
- dispatcher.interrupt();
- pool.shutdown();
- log.info("GrpcRetryer shutdown......");
+ initDispatcher(dispatcher);
}
- public void start() throws Exception {
- dispatcher.start();
- log.info("GrpcRetryer started......");
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
index 14ff727df..891d09901 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/retry/HttpRetryer.java
@@ -19,10 +19,10 @@ package
org.apache.eventmesh.runtime.core.protocol.http.retry;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.runtime.boot.EventMeshHTTPServer;
+import org.apache.eventmesh.runtime.core.protocol.AbstractRetryer;
import org.apache.eventmesh.runtime.core.protocol.DelayRetryable;
import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -33,7 +33,7 @@ import org.slf4j.LoggerFactory;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class HttpRetryer {
+public class HttpRetryer extends AbstractRetryer {
private final Logger retryLogger = LoggerFactory.getLogger("retry");
@@ -43,20 +43,16 @@ public class HttpRetryer {
this.eventMeshHTTPServer = eventMeshHTTPServer;
}
- private final DelayQueue<DelayRetryable> failed = new DelayQueue<>();
-
- private ThreadPoolExecutor pool;
-
- private Thread dispatcher;
-
+ @Override
public void pushRetry(DelayRetryable delayRetryable) {
- if (failed.size() >=
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshServerRetryBlockQSize())
{
+ if (retrys.size() >=
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshServerRetryBlockQSize())
{
retryLogger.error("[RETRY-QUEUE] is full!");
return;
}
- failed.offer(delayRetryable);
+ retrys.offer(delayRetryable);
}
+ @Override
public void init() {
pool = new
ThreadPoolExecutor(eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshServerRetryThreadNum(),
eventMeshHTTPServer.getEventMeshHttpConfiguration().getEventMeshServerRetryThreadNum(),
@@ -66,52 +62,7 @@ public class HttpRetryer {
new EventMeshThreadFactory("http-retry", true,
Thread.NORM_PRIORITY),
new ThreadPoolExecutor.AbortPolicy());
- dispatcher = new Thread(() -> {
- try {
- DelayRetryable retryObj;
- while (!Thread.currentThread().isInterrupted() && (retryObj =
failed.take()) != null) {
- final DelayRetryable delayRetryable = retryObj;
- pool.execute(() -> {
- try {
- delayRetryable.retry();
- if (retryLogger.isDebugEnabled()) {
- retryLogger.debug("retryObj : {}",
delayRetryable);
- }
- } catch (Exception e) {
- retryLogger.error("http-retry-dispatcher error!",
e);
- }
- });
- }
- } catch (Exception e) {
- if (e instanceof InterruptedException) {
- Thread.currentThread().interrupt();
- }
- retryLogger.error("http-retry-dispatcher error!", e);
- }
- }, "http-retry-dispatcher");
- dispatcher.setDaemon(true);
- log.info("HttpRetryer inited......");
- }
-
- public int size() {
- return failed.size();
+ initDispatcher(dispatcher);
}
- /**
- * Get failed queue, this method is just used for metrics.
- */
- public DelayQueue<DelayRetryable> getFailedQueue() {
- return failed;
- }
-
- public void shutdown() {
- dispatcher.interrupt();
- pool.shutdown();
- log.info("HttpRetryer shutdown......");
- }
-
- public void start() throws Exception {
- dispatcher.start();
- log.info("HttpRetryer started......");
- }
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
index 1975a0901..bacfdd71b 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/tcp/client/session/retry/EventMeshTcpRetryer.java
@@ -20,13 +20,14 @@ package
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.retry;
import org.apache.eventmesh.common.EventMeshThreadFactory;
import org.apache.eventmesh.common.protocol.SubscriptionType;
import org.apache.eventmesh.runtime.boot.EventMeshTCPServer;
+import org.apache.eventmesh.runtime.core.protocol.AbstractRetryer;
+import org.apache.eventmesh.runtime.core.protocol.DelayRetryable;
import org.apache.eventmesh.runtime.core.protocol.RetryContext;
import
org.apache.eventmesh.runtime.core.protocol.tcp.client.session.push.DownStreamMsgContext;
import org.apache.eventmesh.runtime.util.EventMeshUtil;
import org.apache.eventmesh.runtime.util.ThreadPoolHelper;
import java.util.concurrent.ArrayBlockingQueue;
-import java.util.concurrent.DelayQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@@ -34,12 +35,10 @@ import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@Slf4j
-public class EventMeshTcpRetryer {
+public class EventMeshTcpRetryer extends AbstractRetryer {
private EventMeshTCPServer eventMeshTCPServer;
- private final DelayQueue<RetryContext> retrys = new DelayQueue<>();
-
private final ThreadPoolExecutor pool = new ThreadPoolExecutor(3,
3,
60000,
@@ -47,8 +46,6 @@ public class EventMeshTcpRetryer {
new EventMeshThreadFactory("eventMesh-tcp-retry", true),
new ThreadPoolExecutor.AbortPolicy());
- private Thread dispatcher;
-
public EventMeshTcpRetryer(EventMeshTCPServer eventMeshTCPServer) {
this.eventMeshTCPServer = eventMeshTCPServer;
}
@@ -61,7 +58,9 @@ public class EventMeshTcpRetryer {
this.eventMeshTCPServer = eventMeshTCPServer;
}
- public void pushRetry(RetryContext retryContext) {
+ @Override
+ public void pushRetry(DelayRetryable delayRetryable) {
+ RetryContext retryContext = (RetryContext) delayRetryable;
if (retrys.size() >=
eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshTcpMsgRetryQueueSize())
{
log.error("pushRetry fail, retrys is too much,allow max
retryQueueSize:{}, retryTimes:{}, seq:{}, bizSeq:{}",
eventMeshTCPServer.getEventMeshTCPConfiguration().getEventMeshTcpMsgRetryQueueSize(),
retryContext.retryTimes,
@@ -88,43 +87,9 @@ public class EventMeshTcpRetryer {
EventMeshUtil.getMessageBizSeq(retryContext.event));
}
+ @Override
public void init() {
- dispatcher = new Thread(() -> {
- try {
- RetryContext retryContext;
- while ((retryContext = retrys.take()) != null) {
- final RetryContext retryCtx = retryContext;
- pool.execute(() -> {
- try {
- retryCtx.retry();
- } catch (Exception e) {
- log.error("retry-dispatcher error!", e);
- }
- });
- }
- } catch (Exception e) {
- if (e instanceof InterruptedException) {
- Thread.currentThread().interrupt();
- }
- log.error("retry-dispatcher error!", e);
- }
- }, "retry-dispatcher");
- dispatcher.setDaemon(true);
- log.info("EventMeshTcpRetryer inited......");
- }
-
- public void start() throws Exception {
- dispatcher.start();
- log.info("EventMeshTcpRetryer started......");
- }
-
- public void shutdown() {
- pool.shutdown();
- log.info("EventMeshTcpRetryer shutdown......");
- }
-
- public int getRetrySize() {
- return retrys.size();
+ initDispatcher(dispatcher);
}
public void printRetryThreadPoolState() {
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/grpc/EventMeshGrpcMonitor.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/grpc/EventMeshGrpcMonitor.java
index a6d5c3f3b..e7ea5d9e3 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/grpc/EventMeshGrpcMonitor.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/grpc/EventMeshGrpcMonitor.java
@@ -64,7 +64,7 @@ public class EventMeshGrpcMonitor {
scheduleTask = scheduler.scheduleAtFixedRate(() -> {
grpcSummaryMetrics.refreshTpsMetrics(SCHEDULE_PERIOD_MILLS);
grpcSummaryMetrics.clearAllMessageCounter();
-
grpcSummaryMetrics.setRetrySize(eventMeshGrpcServer.getGrpcRetryer().size());
+
grpcSummaryMetrics.setRetrySize(eventMeshGrpcServer.getGrpcRetryer().getRetrySize());
grpcSummaryMetrics.setSubscribeTopicNum(eventMeshGrpcServer.getConsumerManager().getAllConsumerTopic().size());
}, DELAY_MILLS, SCHEDULE_PERIOD_MILLS, TimeUnit.MILLISECONDS);
}
diff --git
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
index dc39b23d9..b1e133bac 100644
---
a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
+++
b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/metrics/http/HTTPMetricsServer.java
@@ -51,7 +51,7 @@ public class HTTPMetricsServer {
eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor(),
eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor(),
eventMeshHTTPServer.getHttpThreadPoolGroup().getPushMsgExecutor(),
- eventMeshHTTPServer.getHttpRetryer().getFailedQueue());
+ eventMeshHTTPServer.getHttpRetryer().getRetryQueue());
init();
}
@@ -171,7 +171,7 @@ public class HTTPMetricsServer {
eventMeshHTTPServer.getHttpThreadPoolGroup().getBatchMsgExecutor().getQueue().size(),
eventMeshHTTPServer.getHttpThreadPoolGroup().getSendMsgExecutor().getQueue().size(),
eventMeshHTTPServer.getHttpThreadPoolGroup().getPushMsgExecutor().getQueue().size(),
- eventMeshHTTPServer.getHttpRetryer().size());
+ eventMeshHTTPServer.getHttpRetryer().getRetrySize());
}
if (log.isInfoEnabled()) {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
