Spring Integration中多ThreadStatePropagationChannelInterceptor冲突问题求助

问题根源 #

ThreadStatePropagationChannelInterceptor依赖的MessageWithThreadState仅支持存储单个状态对象，多个同类型拦截器注册时，后注册的实例会尝试将前一个拦截器存入的状态（比如SecurityContext）强制转换为自身所需类型，触发ClassCastException，错误日志如下：

2023-09-07 09:51:16,643 [task-scheduler-6] ERROR org.springframework.integration.handler.LoggingHandler - org.springframework.messaging.MessagingException: nested exception is java.lang.ClassCastException: class org.springframework.security.authentication.UsernamePasswordAuthenticationToken cannot be cast to class com....origin.OriginData (org.springframework.security.authentication.UsernamePasswordAuthenticationToken and com....origin.OriginData are in unnamed module of loader org.apache.catalina.loader.ParallelWebappClassLoader @3070f3e6)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:426)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$4(AbstractPollingEndpoint.java:347)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.lambda$execute$0(ErrorHandlingTaskExecutor.java:57)
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
    at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:55)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.lambda$createPoller$5(AbstractPollingEndpoint.java:340)
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
    at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:95)
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
    at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
    at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.ClassCastException: class org.springframework.security.authentication.UsernamePasswordAuthenticationToken cannot be cast to class com....origin.OriginData (org.springframework.security.authentication.UsernamePasswordAuthenticationToken and com....origin.OriginData are in unnamed module of loader org.apache.catalina.loader.ParallelWebappClassLoader @3070f3e6)
    at com....origin.OriginDataPropagationChannelInterceptor.populatePropagatedContext(OriginDataPropagationChannelInterceptor.java:25)
    at org.springframework.integration.channel.interceptor.ThreadStatePropagationChannelInterceptor.postReceive(ThreadStatePropagationChannelInterceptor.java:72)
    at org.springframework.integration.channel.AbstractMessageChannel$ChannelInterceptorList.postReceive(AbstractMessageChannel.java:522)
    at org.springframework.integration.channel.AbstractPollableChannel.receive(AbstractPollableChannel.java:108)
    at org.springframework.integration.endpoint.PollingConsumer.receiveMessage(PollingConsumer.java:217)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:443)
    at org.springframework.integration.endpoint.AbstractPollingEndpoint.pollForMessage(AbstractPollingEndpoint.java:412)
    ... 13 more

可行解决方案 #

1. 自定义聚合型传播拦截器 #

创建一个统一的拦截器，将SecurityContext和所有自定义ThreadLocal数据打包成一个复合状态对象，统一处理存储与恢复逻辑，避免多个拦截器争抢状态存储位：

// 定义复合状态类，包含所有需要传播的ThreadLocal数据
public class CompositeThreadState {
    private final Authentication securityContext;
    private final OriginData originData;

    public CompositeThreadState(Authentication securityContext, OriginData originData) {
        this.securityContext = securityContext;
        this.originData = originData;
    }

    // getter方法
    public Authentication getSecurityContext() { return securityContext; }
    public OriginData getOriginData() { return originData; }
}

// 自定义聚合拦截器
public class CompositeThreadStatePropagationInterceptor extends ThreadStatePropagationChannelInterceptor {

    @Override
    protected Object obtainPropagatedState(Message<?> message) {
        // 收集当前线程所有需要传播的状态
        Authentication auth = SecurityContextHolder.getContext().getAuthentication();
        OriginData originData = OriginDataHolder.get();
        return new CompositeThreadState(auth, originData);
    }

    @Override
    protected void populatePropagatedContext(Object state) {
        CompositeThreadState compositeState = (CompositeThreadState) state;
        // 恢复SecurityContext
        SecurityContextHolder.getContext().setAuthentication(compositeState.getSecurityContext());
        // 恢复自定义ThreadLocal数据
        OriginDataHolder.set(compositeState.getOriginData());
    }
}

替换原有SecurityContextPropagationChannelInterceptor和自定义拦截器，仅注册这一个聚合拦截器即可。

2. 通过MessageHeaders传递自定义数据 #

如果自定义数据不需要严格绑定ThreadLocal生命周期，可直接将数据放入消息头，接收端从消息头取出后再设置到ThreadLocal：

// 发送端拦截器：将自定义数据放入消息头
public class OriginDataHeaderInterceptor implements ChannelInterceptor {
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        OriginData originData = OriginDataHolder.get();
        return MessageBuilder.fromMessage(message)
                .setHeader("X-ORIGIN-DATA", originData)
                .build();
    }
}

// 接收端拦截器：从消息头取出数据并设置到ThreadLocal
public class OriginDataHeaderRestoreInterceptor implements ChannelInterceptor {
    @Override
    public Message<?> postReceive(Message<?> message, MessageChannel channel) {
        if (message != null) {
            OriginData originData = (OriginData) message.getHeaders().get("X-ORIGIN-DATA");
            OriginDataHolder.set(originData);
        }
        return message;
    }
}

该方式不依赖ThreadStatePropagationChannelInterceptor机制，完全独立，不会与SecurityContext传播冲突。

3. 扩展多状态支持的Message包装类 #

自定义支持多状态存储的MessageWithMultiThreadState，用Map保存不同类型状态，让自定义拦截器识别并操作自身状态项：

// 自定义多状态消息包装类
public class MessageWithMultiThreadState<T> extends MessageWithThreadState<T> {
    private final Map<Class<?>, Object> multiState;

    public MessageWithMultiThreadState(Message<T> delegate, Map<Class<?>, Object> multiState) {
        super(delegate, null); // 忽略父类单个状态
        this.multiState = multiState;
    }

    public <S> S getState(Class<S> stateType) {
        return stateType.cast(multiState.get(stateType));
    }
}

// 修改自定义拦截器逻辑
public class OriginDataPropagationInterceptor extends ThreadStatePropagationChannelInterceptor {
    @Override
    protected Object obtainPropagatedState(Message<?> message) {
        Map<Class<?>, Object> stateMap = new HashMap<>();
        if (message instanceof MessageWithMultiThreadState) {
            MessageWithMultiThreadState<?> multiStateMsg = (MessageWithMultiThreadState<?>) message;
            // 保留已有的状态
            stateMap.put(Authentication.class, multiStateMsg.getState(Authentication.class));
        } else {
            // 初始化状态
            stateMap.put(Authentication.class, SecurityContextHolder.getContext().getAuthentication());
        }
        // 添加自定义状态
        stateMap.put(OriginData.class, OriginDataHolder.get());
        return stateMap;
    }

    @Override
    protected Message<?> wrapMessage(Message<?> message, Object state) {
        return new MessageWithMultiThreadState<>(message, (Map<Class<?>, Object>) state);
    }

    @Override
    protected void populatePropagatedContext(Object state) {
        Map<Class<?>, Object> stateMap = (Map<Class<?>, Object>) state;
        OriginData originData = stateMap.get(OriginData.class);
        OriginDataHolder.set(originData);
    }
}

同时需要修改SecurityContext传播拦截器逻辑，使其识别多状态容器。该方式灵活性高，但需修改多个拦截器实现，复杂度较高。

内容的提问来源于stack exchange，提问作者Piers Geyman