fix: 修复工作流同步执行并发监听异常
- 使用写时复制集合保证事件分发期间安全增删监听器 - 按工作流实例路由同步结果并补充并发回归测试
This commit is contained in:
@@ -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;
|
||||
|
||||
/**
|
||||
* 管理工作流执行过程中的事件、输出及错误监听器。
|
||||
*
|
||||
* <p>监听器的注册与移除频率远低于事件分发频率，使用写时复制集合保证分发过程可以无锁遍历，
|
||||
* 同时允许其他线程安全地注册或移除监听器。</p>
|
||||
*/
|
||||
public class EventManager {
|
||||
|
||||
private static final Logger log = LoggerFactory.getLogger(EventManager.class);
|
||||
|
||||
protected final Map<Class<?>, List<ChainEventListener>> eventListeners = new ConcurrentHashMap<>();
|
||||
protected final List<ChainOutputListener> outputListeners = Collections.synchronizedList(new ArrayList<>());
|
||||
protected final List<ChainErrorListener> chainErrorListeners = Collections.synchronizedList(new ArrayList<>());
|
||||
protected final List<NodeErrorListener> nodeErrorListeners = Collections.synchronizedList(new ArrayList<>());
|
||||
protected final List<ChainOutputListener> outputListeners = new CopyOnWriteArrayList<>();
|
||||
protected final List<ChainErrorListener> chainErrorListeners = new CopyOnWriteArrayList<>();
|
||||
protected final List<NodeErrorListener> nodeErrorListeners = new CopyOnWriteArrayList<>();
|
||||
|
||||
/**
|
||||
* ---------- 通用事件监听器 ----------
|
||||
*/
|
||||
public void addEventListener(Class<? extends Event> 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) {
|
||||
|
||||
@@ -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<String, CompletableFuture<Map<String, Object>>> 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<Map<String, Object>> 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<String, Object> execResult = state.getExecuteResult();
|
||||
future.complete(execResult != null ? execResult : Collections.emptyMap());
|
||||
CompletableFuture<Map<String, Object>> existing = pendingExecutions.putIfAbsent(stateInstanceId, future);
|
||||
if (existing != null) {
|
||||
throw new IllegalStateException("Duplicate pending chain execution: " + stateInstanceId);
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
ChainErrorListener errorListener = (error, c) -> {
|
||||
if (c.getStateInstanceId().equals(stateInstanceId)) {
|
||||
future.completeExceptionally(error);
|
||||
}
|
||||
};
|
||||
|
||||
try {
|
||||
this.addEventListener(listener);
|
||||
this.addErrorListener(errorListener);
|
||||
chain.start(variables);
|
||||
Map<String, Object> 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<Map<String, Object>> 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<String, Object> 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<Map<String, Object>> future = pendingExecutions.get(chain.getStateInstanceId());
|
||||
if (future != null) {
|
||||
future.completeExceptionally(error);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,112 @@
|
||||
/**
|
||||
* Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
|
||||
* <p>
|
||||
* 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
|
||||
* <p>
|
||||
* http://www.gnu.org/licenses/lgpl-3.0.txt
|
||||
* <p>
|
||||
* 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<Future<Map<String, Object>>> 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<Map<String, Object>> 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;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
/**
|
||||
* Copyright (c) 2025-2026, Michael Yang 杨福海 (fuhai999@gmail.com).
|
||||
* <p>
|
||||
* 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
|
||||
* <p>
|
||||
* http://www.gnu.org/licenses/lgpl-3.0.txt
|
||||
* <p>
|
||||
* 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<Throwable> 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());
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user