From 7e59f0e638c8dba31d1a6e55c4bf88cd1052c844 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Thu, 23 Jul 2026 11:52:50 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=B7=A5=E4=BD=9C?= =?UTF-8?q?=E6=B5=81=E5=90=8C=E6=AD=A5=E6=89=A7=E8=A1=8C=E5=B9=B6=E5=8F=91?= =?UTF-8?q?=E7=9B=91=E5=90=AC=E5=BC=82=E5=B8=B8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用写时复制集合保证事件分发期间安全增删监听器 - 按工作流实例路由同步结果并补充并发回归测试 --- .../flow/core/chain/EventManager.java | 18 ++- .../core/chain/runtime/ChainExecutor.java | 84 +++++++++---- .../test/ChainExecutorConcurrencyTest.java | 112 ++++++++++++++++++ .../test/EventManagerConcurrencyTest.java | 91 ++++++++++++++ 4 files changed, 278 insertions(+), 27 deletions(-) create mode 100644 easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java create mode 100644 easy-agents-flow/src/test/java/com/easyagents/flow/core/test/EventManagerConcurrencyTest.java diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/EventManager.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/EventManager.java index 50787cd..3955d38 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/EventManager.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/EventManager.java @@ -20,23 +20,31 @@ import com.easyagents.flow.core.chain.listener.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import java.util.*; +import java.util.List; +import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; +/** + * 管理工作流执行过程中的事件、输出及错误监听器。 + * + *

监听器的注册与移除频率远低于事件分发频率，使用写时复制集合保证分发过程可以无锁遍历， + * 同时允许其他线程安全地注册或移除监听器。

+ */ public class EventManager { private static final Logger log = LoggerFactory.getLogger(EventManager.class); protected final Map, List> eventListeners = new ConcurrentHashMap<>(); - protected final List outputListeners = Collections.synchronizedList(new ArrayList<>()); - protected final List chainErrorListeners = Collections.synchronizedList(new ArrayList<>()); - protected final List nodeErrorListeners = Collections.synchronizedList(new ArrayList<>()); + protected final List outputListeners = new CopyOnWriteArrayList<>(); + protected final List chainErrorListeners = new CopyOnWriteArrayList<>(); + protected final List nodeErrorListeners = new CopyOnWriteArrayList<>(); /** * ---------- 通用事件监听器 ---------- */ public void addEventListener(Class eventClass, ChainEventListener listener) { - eventListeners.computeIfAbsent(eventClass, k -> Collections.synchronizedList(new ArrayList<>())).add(listener); + eventListeners.computeIfAbsent(eventClass, key -> new CopyOnWriteArrayList<>()).add(listener); } public void addEventListener(ChainEventListener listener) { diff --git a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java index 3f6bdb2..a63bfa3 100644 --- a/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java +++ b/easy-agents-flow/src/main/java/com/easyagents/flow/core/chain/runtime/ChainExecutor.java @@ -44,6 +44,9 @@ public class ChainExecutor { private final NodeStateRepository nodeStateRepository; private final TriggerScheduler triggerScheduler; private final EventManager eventManager = new EventManager(); + /** 等待同步执行结果的任务，按工作流实例 ID 进行常量时间路由。 */ + private final ConcurrentMap>> pendingExecutions = + new ConcurrentHashMap<>(); public ChainExecutor(ChainDefinitionRepository definitionRepository , ChainStateRepository chainStateRepository @@ -53,7 +56,7 @@ public class ChainExecutor { this.chainStateRepository = chainStateRepository; this.nodeStateRepository = nodeStateRepository; this.triggerScheduler = ChainRuntime.triggerScheduler(); - this.triggerScheduler.registerConsumer(this::accept); + registerRuntimeCallbacks(); } @@ -65,7 +68,7 @@ public class ChainExecutor { this.chainStateRepository = chainStateRepository; this.nodeStateRepository = nodeStateRepository; this.triggerScheduler = triggerScheduler; - this.triggerScheduler.registerConsumer(this::accept); + registerRuntimeCallbacks(); } @@ -79,26 +82,12 @@ public class ChainExecutor { String stateInstanceId = chain.getStateInstanceId(); CompletableFuture> future = new CompletableFuture<>(); - ChainEventListener listener = (event, c) -> { - if (event instanceof ChainStatusChangeEvent) { - if (((ChainStatusChangeEvent) event).getStatus().isTerminal() - && c.getStateInstanceId().equals(stateInstanceId)) { - ChainState state = chainStateRepository.load(stateInstanceId); - Map execResult = state.getExecuteResult(); - future.complete(execResult != null ? execResult : Collections.emptyMap()); - } - } - }; - - ChainErrorListener errorListener = (error, c) -> { - if (c.getStateInstanceId().equals(stateInstanceId)) { - future.completeExceptionally(error); - } - }; + CompletableFuture> existing = pendingExecutions.putIfAbsent(stateInstanceId, future); + if (existing != null) { + throw new IllegalStateException("Duplicate pending chain execution: " + stateInstanceId); + } try { - this.addEventListener(listener); - this.addErrorListener(errorListener); chain.start(variables); Map result = future.get(timeout, unit); clearDefaultStates(result); @@ -114,8 +103,59 @@ public class ChainExecutor { future.cancel(true); throw new RuntimeException("Execution failed", e.getCause()); } finally { - this.removeEventListener(listener); - this.removeErrorListener(errorListener); + pendingExecutions.remove(stateInstanceId, future); + } + } + + /** + * 注册工作流调度和同步结果路由回调。 + */ + private void registerRuntimeCallbacks() { + eventManager.addEventListener(ChainStatusChangeEvent.class, this::completePendingExecution); + eventManager.addChainErrorListener(this::failPendingExecution); + triggerScheduler.registerConsumer(this::accept); + } + + /** + * 在工作流进入终态时完成对应的同步等待任务。 + * + * @param event 工作流状态事件 + * @param chain 产生事件的工作流实例 + */ + private void completePendingExecution(Event event, Chain chain) { + if (!(event instanceof ChainStatusChangeEvent statusChangeEvent) + || !statusChangeEvent.getStatus().isTerminal()) { + return; + } + + String stateInstanceId = chain.getStateInstanceId(); + CompletableFuture> future = pendingExecutions.get(stateInstanceId); + if (future == null) { + return; + } + + try { + ChainState state = chainStateRepository.load(stateInstanceId); + if (state == null) { + throw new ChainException("Chain state not found: " + stateInstanceId); + } + Map execResult = state.getExecuteResult(); + future.complete(execResult != null ? execResult : Collections.emptyMap()); + } catch (Exception error) { + future.completeExceptionally(error); + } + } + + /** + * 将工作流执行异常传递给对应的同步等待任务。 + * + * @param error 工作流执行异常 + * @param chain 发生异常的工作流实例 + */ + private void failPendingExecution(Throwable error, Chain chain) { + CompletableFuture> future = pendingExecutions.get(chain.getStateInstanceId()); + if (future != null) { + future.completeExceptionally(error); } } diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java new file mode 100644 index 0000000..482736c --- /dev/null +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/ChainExecutorConcurrencyTest.java @@ -0,0 +1,112 @@ +/** + * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com). + *

+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.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.gnu.org/licenses/lgpl-3.0.txt + *

+ * 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 com.easyagents.flow.core.test; + +import com.easyagents.flow.core.chain.ChainDefinition; +import com.easyagents.flow.core.chain.Edge; +import com.easyagents.flow.core.chain.repository.InMemoryChainStateRepository; +import com.easyagents.flow.core.chain.repository.InMemoryNodeStateRepository; +import com.easyagents.flow.core.chain.runtime.ChainExecutor; +import com.easyagents.flow.core.chain.runtime.InMemoryTriggerStore; +import com.easyagents.flow.core.chain.runtime.TriggerScheduler; +import com.easyagents.flow.core.node.EndNode; +import com.easyagents.flow.core.node.StartNode; +import org.junit.Assert; +import org.junit.Test; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +/** + * {@link ChainExecutor} 并发同步执行测试。 + */ +public class ChainExecutorConcurrencyTest { + + /** + * 验证多个同步调用可以通过实例 ID 独立接收执行结果。 + * + * @throws Exception 并发任务执行失败、超时或中断时抛出 + */ + @Test + public void shouldRouteConcurrentSynchronousExecutionResults() throws Exception { + ScheduledExecutorService schedulerPool = Executors.newScheduledThreadPool(2); + ExecutorService workerPool = Executors.newFixedThreadPool(8); + ExecutorService callerPool = Executors.newFixedThreadPool(8); + TriggerScheduler triggerScheduler = new TriggerScheduler( + new InMemoryTriggerStore(), schedulerPool, workerPool, 1000L); + ChainDefinition definition = createDefinition(); + ChainExecutor chainExecutor = new ChainExecutor( + id -> definition, + new InMemoryChainStateRepository(), + new InMemoryNodeStateRepository(), + triggerScheduler); + + int executionCount = 16; + CountDownLatch startGate = new CountDownLatch(1); + List>> executions = new ArrayList<>(executionCount); + + try { + for (int index = 0; index < executionCount; index++) { + executions.add(callerPool.submit(() -> { + startGate.await(); + return chainExecutor.execute(definition.getId(), Collections.emptyMap()); + })); + } + startGate.countDown(); + + for (Future> execution : executions) { + Assert.assertNotNull(execution.get(10, TimeUnit.SECONDS)); + } + } finally { + startGate.countDown(); + callerPool.shutdownNow(); + triggerScheduler.shutdown(); + } + } + + /** + * 创建仅包含开始和结束节点的测试工作流。 + * + * @return 测试工作流定义 + */ + private ChainDefinition createDefinition() { + ChainDefinition definition = new ChainDefinition(); + definition.setId("concurrent-sync-test"); + + StartNode startNode = new StartNode(); + startNode.setId("start"); + EndNode endNode = new EndNode(); + endNode.setId("end"); + + Edge edge = new Edge(); + edge.setId("start-to-end"); + edge.setSource(startNode.getId()); + edge.setTarget(endNode.getId()); + + definition.addNode(startNode); + definition.addNode(endNode); + definition.addEdge(edge); + return definition; + } +} diff --git a/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/EventManagerConcurrencyTest.java b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/EventManagerConcurrencyTest.java new file mode 100644 index 0000000..67279c2 --- /dev/null +++ b/easy-agents-flow/src/test/java/com/easyagents/flow/core/test/EventManagerConcurrencyTest.java @@ -0,0 +1,91 @@ +/** + * Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com). + *

+ * Licensed under the GNU Lesser General Public License (LGPL) ,Version 3.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.gnu.org/licenses/lgpl-3.0.txt + *

+ * 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 com.easyagents.flow.core.test; + +import com.easyagents.flow.core.chain.EventManager; +import com.easyagents.flow.core.chain.event.ChainEndEvent; +import com.easyagents.flow.core.chain.listener.ChainEventListener; +import org.junit.Assert; +import org.junit.Test; + +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; + +/** + * {@link EventManager} 并发行为测试。 + */ +public class EventManagerConcurrencyTest { + + /** + * 验证事件分发期间可以从其他线程移除监听器，当前分发使用稳定快照，后续分发不再调用已移除监听器。 + * + * @throws Exception 等待测试线程超时或中断时抛出 + */ + @Test + public void shouldAllowListenerRemovalDuringEventDispatch() throws Exception { + EventManager eventManager = new EventManager(); + CountDownLatch dispatchStarted = new CountDownLatch(1); + CountDownLatch continueDispatch = new CountDownLatch(1); + AtomicInteger retainedInvocationCount = new AtomicInteger(); + AtomicInteger removedInvocationCount = new AtomicInteger(); + AtomicReference dispatchFailure = new AtomicReference<>(); + + ChainEventListener blockingListener = (event, chain) -> { + dispatchStarted.countDown(); + try { + if (!continueDispatch.await(3, TimeUnit.SECONDS)) { + throw new AssertionError("事件分发等待超时"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new AssertionError("事件分发线程被中断", e); + } + }; + ChainEventListener retainedListener = (event, chain) -> retainedInvocationCount.incrementAndGet(); + ChainEventListener removedListener = (event, chain) -> removedInvocationCount.incrementAndGet(); + eventManager.addEventListener(blockingListener); + eventManager.addEventListener(retainedListener); + eventManager.addEventListener(removedListener); + + Thread dispatchThread = new Thread(() -> { + try { + eventManager.notifyEvent(new ChainEndEvent(null), null); + } catch (Throwable error) { + dispatchFailure.set(error); + } + }, "event-manager-dispatch-test"); + dispatchThread.start(); + + try { + Assert.assertTrue("事件分发应在测试超时前开始", dispatchStarted.await(3, TimeUnit.SECONDS)); + eventManager.removeEventListener(removedListener); + } finally { + continueDispatch.countDown(); + } + + dispatchThread.join(3000L); + Assert.assertFalse("事件分发线程应正常结束", dispatchThread.isAlive()); + Assert.assertNull("并发移除监听器不应中断事件分发", dispatchFailure.get()); + Assert.assertEquals(1, retainedInvocationCount.get()); + Assert.assertEquals("当前事件应按分发开始时的快照通知", 1, removedInvocationCount.get()); + + eventManager.notifyEvent(new ChainEndEvent(null), null); + Assert.assertEquals(2, retainedInvocationCount.get()); + Assert.assertEquals("后续事件不应通知已移除监听器", 1, removedInvocationCount.get()); + } +}