This is an automated email from the ASF dual-hosted git repository.
xiatian 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 87172a943 [ISSUE #4796] Use ThreadPoolFactory to create single thread
executor for SourceWorker (#4798)
87172a943 is described below
commit 87172a94327d3abaff101759a5a21aa0fea11427
Author: Jevin Jiang <[email protected]>
AuthorDate: Thu Mar 21 01:41:34 2024 +0800
[ISSUE #4796] Use ThreadPoolFactory to create single thread executor for
SourceWorker (#4798)
---
.../main/java/org/apache/eventmesh/common/ThreadPoolFactory.java | 5 +++++
.../main/java/org/apache/eventmesh/openconnect/SourceWorker.java | 8 +++++---
2 files changed, 10 insertions(+), 3 deletions(-)
diff --git
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/ThreadPoolFactory.java
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ThreadPoolFactory.java
index 7b1cc6d94..6f9d8a6ba 100644
---
a/eventmesh-common/src/main/java/org/apache/eventmesh/common/ThreadPoolFactory.java
+++
b/eventmesh-common/src/main/java/org/apache/eventmesh/common/ThreadPoolFactory.java
@@ -18,6 +18,7 @@
package org.apache.eventmesh.common;
import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ScheduledExecutorService;
@@ -57,4 +58,8 @@ public abstract class ThreadPoolFactory {
public static ScheduledExecutorService createScheduledExecutor(int core,
ThreadFactory threadFactory) {
return Executors.newScheduledThreadPool(core, threadFactory);
}
+
+ public static ExecutorService createSingleExecutor(final String
threadName) {
+ return Executors.newSingleThreadExecutor(new
EventMeshThreadFactory(threadName));
+ }
}
diff --git
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
index b971d0b98..6dcbcd299 100644
---
a/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
+++
b/eventmesh-openconnect/eventmesh-openconnect-java/src/main/java/org/apache/eventmesh/openconnect/SourceWorker.java
@@ -23,6 +23,7 @@ import org.apache.eventmesh.client.tcp.EventMeshTCPClient;
import org.apache.eventmesh.client.tcp.EventMeshTCPClientFactory;
import org.apache.eventmesh.client.tcp.common.MessageUtils;
import org.apache.eventmesh.client.tcp.conf.EventMeshTCPClientConfig;
+import org.apache.eventmesh.common.ThreadPoolFactory;
import org.apache.eventmesh.common.exception.EventMeshException;
import org.apache.eventmesh.common.protocol.tcp.OPStatus;
import org.apache.eventmesh.common.protocol.tcp.Package;
@@ -56,7 +57,6 @@ import java.util.UUID;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
@@ -87,9 +87,11 @@ public class SourceWorker implements ConnectorWorker {
private volatile RecordOffsetManagement.CommittableOffsets
committableOffsets;
- private final ExecutorService pollService =
Executors.newSingleThreadExecutor();
+ private final ExecutorService pollService =
+
ThreadPoolFactory.createSingleExecutor("eventMesh-sourceWorker-pollService");
- private final ExecutorService startService =
Executors.newSingleThreadExecutor();
+ private final ExecutorService startService =
+
ThreadPoolFactory.createSingleExecutor("eventMesh-sourceWorker-startService");
private final BlockingQueue<ConnectRecord> queue;
private final EventMeshTCPClient<CloudEvent> eventMeshTCPClient;
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
