系统优化

framework-base/src/main/java/com/chelvc/framework/base/config/ThreadPoolConfigurer.java

@@ -35,7 +35,7 @@ public class ThreadPoolConfigurer {
         executor.setKeepAliveSeconds((int) config.getPool().getKeepAlive().getSeconds());
         executor.setWaitForTasksToCompleteOnShutdown(config.getShutdown().isAwaitTermination());
         executor.setAwaitTerminationSeconds((int) config.getShutdown().getAwaitTerminationPeriod().getSeconds());
-        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
+        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
         executor.initialize();
         executor.getThreadPoolExecutor().prestartAllCoreThreads();
         return executor;

framework-base/src/main/java/com/chelvc/framework/base/context/ExecutorContextHolder.java

@@ -5,16 +5,19 @@ import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
 import java.util.function.Consumer;
 import java.util.function.Supplier;
 
 import com.chelvc.framework.base.model.Session;
+import com.chelvc.framework.base.util.ThreadUtils;
 import lombok.NonNull;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.task.TaskExecutor;
+import org.springframework.scheduling.TaskScheduler;
 import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
 
@@ -22,7 +25,7 @@ import org.springframework.util.CollectionUtils;
  * Executor上下文工具类
  *
  * @author Woody
- * @date 2023/4/5
+ * @date 2021/9/8
  */
 @Slf4j
 @Component
@@ -33,22 +36,52 @@ public class ExecutorContextHolder implements ApplicationContextAware {
     private static final CompletableFuture<?>[] EMPTY_FUTURES = new CompletableFuture<?>[0];
 
     /**
-     * 多线程任务执行实例
+     * 任务执行线程池
      */
-    private static Executor EXECUTOR;
+    private static TaskExecutor TASK_EXECUTOR;
+
+    /**
+     * 任务调度器
+     */
+    private static TaskScheduler TASK_SCHEDULER;
+
+    /**
+     * 默认任务执行器
+     */
+    private static final Executor DEFAULT_EXECUTOR = ForkJoinPool.getCommonPoolParallelism() > 1 ?
+            ForkJoinPool.commonPool() : ThreadUtils::run;
 
     @Override
     public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        EXECUTOR = applicationContext.getBean(TaskExecutor.class);
+        TASK_EXECUTOR = applicationContext.getBean(TaskExecutor.class);
+        TASK_SCHEDULER = applicationContext.getBean(TaskScheduler.class);
     }
 
     /**
-     * 获取应用上下文中配置的Executor
+     * 获取默认任务执行器
      *
-     * @return Executor对象
+     * @return 任务执行器实例
      */
-    public static Executor getExecutor() {
-        return Objects.requireNonNull(EXECUTOR, "Executor has not been initialized");
+    public static Executor getDefaultExecutor() {
+        return DEFAULT_EXECUTOR;
+    }
+
+    /**
+     * 获取任务执行线程池
+     *
+     * @return 任务执行线程池实例
+     */
+    public static TaskExecutor getTaskExecutor() {
+        return Objects.requireNonNull(TASK_EXECUTOR, "Task executor has not been initialized");
+    }
+
+    /**
+     * 获取任务调度器
+     *
+     * @return 任务调度器实例
+     */
+    public static TaskScheduler getTaskScheduler() {
+        return Objects.requireNonNull(TASK_SCHEDULER, "Task scheduler has not been initialized");
     }
 
     /**
@@ -58,7 +91,7 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      * @return CompletableFuture实例
      */
     public static CompletableFuture<Void> run(@NonNull Runnable runnable) {
-        return run(EXECUTOR, runnable);
+        return run(getDefaultExecutor(), runnable);
     }
 
     /**
@@ -68,25 +101,19 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      * @param runnable Runnable实例
      * @return CompletableFuture实例
      */
-    public static CompletableFuture<Void> run(Executor executor, @NonNull Runnable runnable) {
-        long id = Thread.currentThread().getId();
+    public static CompletableFuture<Void> run(@NonNull Executor executor, @NonNull Runnable runnable) {
         Session session = SessionContextHolder.getSession(false);
-        Runnable function = () -> {
-            if (id == Thread.currentThread().getId()) {
-                runnable.run();
-            } else {
-                // 如果子、父线程不是同一个线程则设置会话上下文信息
-                SessionContextHolder.setSession(session);
-                try {
-                    runnable.run();
-                } finally {
-                    SessionContextHolder.clearSessionContext();
-                }
-            }
-        };
-        CompletableFuture<Void> future = executor == null ? CompletableFuture.runAsync(function) :
-                CompletableFuture.runAsync(function, executor);
-        return future.exceptionally(e -> {
+        return CompletableFuture.runAsync(
+                () -> {
+                    SessionContextHolder.setSession(session);
+                    try {
+                        runnable.run();
+                    } finally {
+                        SessionContextHolder.clearSessionContext();
+                    }
+                },
+                executor
+        ).exceptionally(e -> {
             log.error(e.getMessage(), e);
             return null;
         });
@@ -100,7 +127,7 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      * @return CompletableFuture实例
      */
     public static <T> CompletableFuture<T> supply(@NonNull Supplier<T> supplier) {
-        return supply(EXECUTOR, supplier);
+        return supply(getDefaultExecutor(), supplier);
     }
 
     /**
@@ -111,25 +138,19 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      * @param <T>      任务结果类型
      * @return CompletableFuture实例
      */
-    public static <T> CompletableFuture<T> supply(Executor executor, @NonNull Supplier<T> supplier) {
-        long id = Thread.currentThread().getId();
+    public static <T> CompletableFuture<T> supply(@NonNull Executor executor, @NonNull Supplier<T> supplier) {
         Session session = SessionContextHolder.getSession(false);
-        Supplier<T> function = () -> {
-            if (id == Thread.currentThread().getId()) {
-                return supplier.get();
-            }
-
-            // 如果子、父线程不是同一个线程则设置会话上下文信息
-            SessionContextHolder.setSession(session);
-            try {
-                return supplier.get();
-            } finally {
-                SessionContextHolder.clearSessionContext();
-            }
-        };
-        CompletableFuture<T> future = executor == null ? CompletableFuture.supplyAsync(function) :
-                CompletableFuture.supplyAsync(function, executor);
-        return future.exceptionally(e -> {
+        return CompletableFuture.supplyAsync(
+                () -> {
+                    SessionContextHolder.setSession(session);
+                    try {
+                        return supplier.get();
+                    } finally {
+                        SessionContextHolder.clearSessionContext();
+                    }
+                },
+                executor
+        ).exceptionally(e -> {
             log.error(e.getMessage(), e);
             return null;
         });
@@ -141,7 +162,7 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      * @param array Runnable实例数组
      */
     public static void execute(@NonNull Runnable... array) {
-        execute(EXECUTOR, array);
+        execute(getDefaultExecutor(), array);
     }
 
     /**
@@ -150,7 +171,7 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      * @param executor 线程执行器
      * @param array    Runnable实例数组
      */
-    public static void execute(Executor executor, @NonNull Runnable... array) {
+    public static void execute(@NonNull Executor executor, @NonNull Runnable... array) {
         if (array != null && array.length == 1) {
             run(executor, array[0]).join();
         } else if (array != null) {
@@ -171,7 +192,7 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      */
     @SuppressWarnings("unchecked")
     public static <T> CompletableFuture<T>[] execute(@NonNull Supplier<T>... array) {
-        return execute(EXECUTOR, array);
+        return execute(getDefaultExecutor(), array);
     }
 
     /**
@@ -182,12 +203,12 @@ public class ExecutorContextHolder implements ApplicationContextAware {
      * @param <T>      任务结果类型
      * @return CompletableFuture实例
      */
-    @SuppressWarnings({"unchecked", "rawtypes"})
-    public static <T> CompletableFuture<T>[] execute(Executor executor, @NonNull Supplier<T>... array) {
+    @SuppressWarnings("unchecked")
+    public static <T> CompletableFuture<T>[] execute(@NonNull Executor executor, @NonNull Supplier<T>... array) {
         if (array == null || array.length == 0) {
             return new CompletableFuture[0];
         }
-        CompletableFuture[] futures = new CompletableFuture[array.length];
+        CompletableFuture<T>[] futures = new CompletableFuture[array.length];
         for (int i = 0; i < array.length; i++) {
             futures[i] = supply(executor, array[i]);
         }

framework-base/src/main/java/com/chelvc/framework/base/context/SessionContextHolder.java

@@ -142,12 +142,60 @@ public class SessionContextHolder implements ServletRequestListener {
     }
 
     /**
-     * 判断当前会话是否已认证
+     * 初始化会话信息
      *
-     * @return true/false
+     * @param session 会话信息
      */
-    public static boolean isAuthenticated() {
-        return Objects.nonNull(getId(false));
+    public static void initializeSession(@NonNull Session session) {
+        Deque<Session> deque = SESSION_CONTEXT.get();
+        deque.poll();
+        deque.push(session);
+    }
+
+    /**
+     * 初始化会话主体
+     *
+     * @param id       主体标识
+     * @param business 业务类型
+     * @param mobile   电话号码
+     * @return 会话信息
+     */
+    public static Session initializeSessionPrincipal(@NonNull Long id, @NonNull String business, String mobile) {
+        return initializeSessionPrincipal(id, business, mobile, true);
+    }
+
+    /**
+     * 初始化会话主体
+     *
+     * @param id       主体标识
+     * @param business 业务类型
+     * @param mobile   电话号码
+     * @param cover    是否覆盖已认证主体信息
+     * @return 会话信息
+     */
+    public static Session initializeSessionPrincipal(@NonNull Long id, @NonNull String business, String mobile,
+                                                     boolean cover) {
+        Session session = null;
+        Deque<Session> deque = null;
+        if (cover || ObjectUtils.ifNull(session = (deque = SESSION_CONTEXT.get()).peek(), Session::getId) == null) {
+            session = Session.builder().id(id).business(business).mobile(mobile)
+                    .host(ObjectUtils.ifNull(session, Session::getHost))
+                    .device(ObjectUtils.ifNull(session, Session::getDevice))
+                    .tenant(ObjectUtils.ifNull(session, Session::getTenant))
+                    .initial(ObjectUtils.ifNull(session, Session::getInitial))
+                    .channel(ObjectUtils.ifNull(session, Session::getChannel))
+                    .platform(ObjectUtils.ifNull(session, Session::getPlatform))
+                    .terminal(ObjectUtils.ifNull(session, Session::getTerminal))
+                    .version(ObjectUtils.ifNull(session, Session::getVersion))
+                    .sequence(ObjectUtils.ifNull(session, Session::getSequence))
+                    .signature(ObjectUtils.ifNull(session, Session::getSignature)).build();
+            if (deque == null) {
+                deque = SESSION_CONTEXT.get();
+            }
+            deque.poll();
+            deque.push(session);
+        }
+        return session;
     }
 
     /**
@@ -397,6 +445,24 @@ public class SessionContextHolder implements ServletRequestListener {
         return ObjectUtils.ifNull(getSession(requireSession), Session::getSequence);
     }
 
+    /**
+     * 判断当前会话是否存在手机号
+     *
+     * @return true/false
+     */
+    public static boolean hasMobile() {
+        return StringUtils.nonEmpty(getMobile(false));
+    }
+
+    /**
+     * 判断当前会话是否已认证
+     *
+     * @return true/false
+     */
+    public static boolean isAuthenticated() {
+        return Objects.nonNull(getId(false));
+    }
+
     /**
      * 获取缓存对象
      *
@@ -689,22 +755,6 @@ public class SessionContextHolder implements ServletRequestListener {
         return StringUtils.ifEmpty(ObjectUtils.ifNull(getRequest(), request -> request.getParameter(name)), adapter);
     }
 
-    /**
-     * 初始化会话主体
-     *
-     * @param id       主体标识
-     * @param business 业务类型
-     * @param mobile   电话号码
-     */
-    public static void initializeSessionPrincipal(@NonNull Long id, @NonNull String business, String mobile) {
-        Session current = Objects.requireNonNull(SESSION_CONTEXT.get().poll(), "Session has not been initialized");
-        Session session = Session.builder().id(id).host(current.getHost()).mobile(mobile).device(current.getDevice())
-                .tenant(current.getTenant()).initial(current.getInitial()).channel(current.getChannel())
-                .platform(current.getPlatform()).terminal(current.getTerminal()).version(current.getVersion())
-                .business(business).sequence(current.getSequence()).signature(current.getSignature()).build();
-        SESSION_CONTEXT.get().push(session);
-    }
-
     /**
      * 清空上下文信息
      */

framework-dubbo/src/main/java/com/chelvc/framework/dubbo/interceptor/DubboConsumerInterceptor.java

@@ -23,6 +23,11 @@ import org.apache.dubbo.rpc.RpcInvocation;
  */
 @Activate(group = CommonConstants.CONSUMER)
 public class DubboConsumerInterceptor implements Filter {
+    /**
+     * 用户会话上下文名称
+     */
+    public static final String SESSION_CONTEXT_NAME = Session.class.getSimpleName().toUpperCase();
+
     /**
      * 查找接口对象适配器
      *
@@ -51,7 +56,7 @@ public class DubboConsumerInterceptor implements Filter {
     public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
         // 初始化会话信息
         RpcContext.getClientAttachment().setObjectAttachment(
-                Session.class.getName(), SessionContextHolder.getSession(false)
+                SESSION_CONTEXT_NAME, SessionContextHolder.getSession(false)
         );
 
         // 查找对象适配器

framework-dubbo/src/main/java/com/chelvc/framework/dubbo/interceptor/DubboProviderInterceptor.java

@@ -23,41 +23,35 @@ import org.apache.dubbo.rpc.RpcException;
  */
 @Slf4j
 @Activate(group = CommonConstants.PROVIDER)
-public class DubboProviderInterceptor implements Filter, Filter.Listener {
+public class DubboProviderInterceptor implements Filter {
     @Override
     public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
         // 初始化会话信息
-        Session session = (Session) RpcContext.getServerAttachment().getObjectAttachment(Session.class.getName());
+        Session session = (Session) RpcContext.getServerAttachment()
+                .getObjectAttachment(DubboConsumerInterceptor.SESSION_CONTEXT_NAME);
         if (log.isDebugEnabled()) {
             log.debug("[{}] [{}]: {}", invocation.getTargetServiceUniqueName(), invocation.getMethodName(), session);
         }
         SessionContextHolder.setSession(session);
 
         // 调用目标接口
-        Result result = invoker.invoke(invocation);
-        Object value = result.getValue();
-        if (value == null) {
-            return result;
-        }
+        try {
+            Result result = invoker.invoke(invocation);
+            Object value = result.getValue();
+            if (value == null) {
+                return result;
+            }
 
-        // 对象包装
-        DubboObjectAdapterWrapper<Object, ?> wrapper = DubboObjectAdapterHolder.lookupObjectAdapter(value.getClass());
-        if (wrapper == null) {
-            return result;
+            // 对象包装
+            DubboObjectAdapterWrapper<Object, ?> wrapper =
+                    DubboObjectAdapterHolder.lookupObjectAdapter(value.getClass());
+            if (wrapper == null) {
+                return result;
+            }
+            value = wrapper.getAdapter().wrap(result.getValue());
+            return AsyncRpcResult.newDefaultAsyncResult(value, invocation);
+        } finally {
+            SessionContextHolder.clearSessionContext();
         }
-        value = wrapper.getAdapter().wrap(result.getValue());
-        return AsyncRpcResult.newDefaultAsyncResult(value, invocation);
-    }
-
-    @Override
-    public void onResponse(Result appResponse, Invoker<?> invoker, Invocation invocation) {
-        // 清理上下文
-        SessionContextHolder.clearSessionContext();
-    }
-
-    @Override
-    public void onError(Throwable t, Invoker<?> invoker, Invocation invocation) {
-        // 清理上下文
-        SessionContextHolder.clearSessionContext();
     }
 }

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/config/MessageConverterWrapper.java

@@ -1,112 +0,0 @@
-package com.chelvc.framework.rocketmq.config;
-
-import com.chelvc.framework.base.context.JacksonContextHolder;
-import com.chelvc.framework.base.context.SessionContextHolder;
-import com.chelvc.framework.base.model.Session;
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.EqualsAndHashCode;
-import lombok.Getter;
-import lombok.NoArgsConstructor;
-import lombok.NonNull;
-import lombok.ToString;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.MessageHeaders;
-import org.springframework.messaging.converter.MessageConverter;
-import org.springframework.messaging.support.GenericMessage;
-
-/**
- * 消息转换包装器
- *
- * @author Woody
- * @date 2023/4/5
- */
-public class MessageConverterWrapper implements MessageConverter {
-    /**
-     * 消息转换包装目标实例
-     */
-    private final MessageConverter target;
-
-    /**
-     * 应用上下文是否已初始化
-     */
-    private volatile boolean applicationContextInitialized;
-
-    public MessageConverterWrapper(@NonNull MessageConverter target) {
-        this(target, true);
-    }
-
-    public MessageConverterWrapper(@NonNull MessageConverter target, boolean applicationContextInitialized) {
-        this.target = target;
-        this.applicationContextInitialized = applicationContextInitialized;
-    }
-
-    /**
-     * 初始化应用上下文完成
-     */
-    public void initializeApplicationContextCompleted() {
-        synchronized (this) {
-            this.applicationContextInitialized = true;
-            this.notifyAll();
-        }
-    }
-
-    @Override
-    public Object fromMessage(Message<?> message, Class<?> targetClass) {
-        // 等待应用上下文初始化完成
-        if (!this.applicationContextInitialized) {
-            synchronized (this) {
-                if (!this.applicationContextInitialized) {
-                    try {
-                        this.wait();
-                    } catch (InterruptedException ignored) {
-                    }
-                }
-            }
-        }
-
-        // 拆包消息
-        MessageWrapper wrapper = (MessageWrapper) this.target.fromMessage(message, MessageWrapper.class);
-        if (wrapper == null) {
-            return null;
-        }
-        Object payload = wrapper.getTarget();
-        SessionContextHolder.setSession(wrapper.getSession());
-        if (payload == null || (payload instanceof String && targetClass == String.class)) {
-            return payload;
-        }
-        if (!(payload instanceof String)) {
-            payload = JacksonContextHolder.serialize(payload);
-        }
-        return this.target.fromMessage(new GenericMessage<>(payload), targetClass);
-    }
-
-    @Override
-    public Message<?> toMessage(Object payload, MessageHeaders headers) {
-        Object target = JacksonContextHolder.serialize(payload);
-        Session session = SessionContextHolder.getSession(false);
-        MessageWrapper wrapper = MessageWrapper.builder().target(target).session(session).build();
-        return this.target.toMessage(wrapper, headers);
-    }
-
-    /**
-     * 消息包装类
-     */
-    @Getter
-    @Builder
-    @ToString
-    @EqualsAndHashCode
-    @NoArgsConstructor
-    @AllArgsConstructor
-    public static class MessageWrapper {
-        /**
-         * 目标对象
-         */
-        private Object target;
-
-        /**
-         * 用户会话
-         */
-        private Session session;
-    }
-}

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/config/RocketMQConfigurer.java

@@ -1,16 +1,10 @@
 package com.chelvc.framework.rocketmq.config;
 
-import lombok.RequiredArgsConstructor;
 import org.apache.rocketmq.client.log.ClientLogger;
-import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
-import org.springframework.context.ApplicationContext;
-import org.springframework.context.annotation.Bean;
+import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.Primary;
-import org.springframework.messaging.converter.MessageConverter;
 
 /**
  * RocketMQ配置类
@@ -19,35 +13,16 @@ import org.springframework.messaging.converter.MessageConverter;
  * @date 2023/4/5
  */
 @Configuration
-@RequiredArgsConstructor(onConstructor = @__(@Autowired))
-public class RocketMQConfigurer implements ApplicationRunner {
+public class RocketMQConfigurer implements BeanPostProcessor {
     static {
         System.setProperty(ClientLogger.CLIENT_LOG_USESLF4J, "true");
     }
 
-    private final ApplicationContext applicationContext;
-
-    @Bean
-    @Primary
-    public RocketMQMessageConverter rocketMQMessageConverter() {
-        return new RocketMQMessageConverter() {
-            private final MessageConverter messageConverter;
-
-            {
-                // 重置消息转换类，用于封装请求/认证上下文信息
-                this.messageConverter = new MessageConverterWrapper(super.getMessageConverter(), false);
-            }
-
-            @Override
-            public MessageConverter getMessageConverter() {
-                return this.messageConverter;
-            }
-        };
-    }
-
     @Override
-    public void run(ApplicationArguments args) {
-        RocketMQMessageConverter messageConverter = this.applicationContext.getBean(RocketMQMessageConverter.class);
-        ((MessageConverterWrapper) messageConverter.getMessageConverter()).initializeApplicationContextCompleted();
+    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
+        if (bean instanceof DefaultRocketMQListenerContainer) {
+            return new SessionRocketMQListenerContainer((DefaultRocketMQListenerContainer) bean);
+        }
+        return bean;
     }
 }

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/config/SessionRocketMQListenerContainer.java

@@ -0,0 +1,376 @@
+package com.chelvc.framework.rocketmq.config;
+
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.ParameterizedType;
+import java.lang.reflect.Type;
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+import com.chelvc.framework.base.context.JacksonContextHolder;
+import com.chelvc.framework.base.util.ObjectUtils;
+import com.chelvc.framework.rocketmq.context.RocketMQContextHolder;
+import com.chelvc.framework.rocketmq.interceptor.SessionRocketMQListener;
+import com.chelvc.framework.rocketmq.model.MessageContext;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.AccessChannel;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.annotation.ConsumeMode;
+import org.apache.rocketmq.spring.annotation.MessageModel;
+import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
+import org.apache.rocketmq.spring.annotation.SelectorType;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.apache.rocketmq.spring.core.RocketMQReplyListener;
+import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
+import org.springframework.aop.framework.AopProxyUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.core.MethodParameter;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.converter.SmartMessageConverter;
+import org.springframework.messaging.support.MessageBuilder;
+
+/**
+ * 会话消息监听器容器
+ *
+ * @author Woody
+ * @date 2023/5/18
+ */
+@Slf4j
+public class SessionRocketMQListenerContainer extends DefaultRocketMQListenerContainer {
+    private final DefaultRocketMQListenerContainer container;
+    private Type type;
+    private MethodParameter parameter;
+
+    public SessionRocketMQListenerContainer(DefaultRocketMQListenerContainer container) {
+        this.container = container;
+    }
+
+    /**
+     * 容器初始化
+     */
+    @SuppressWarnings({"unchecked", "rawtypes"})
+    private void initialize() {
+        // 获取目标消息监听器对象类型
+        Class<?> target;
+        if (this.getRocketMQListener() != null) {
+            target = AopProxyUtils.ultimateTargetClass(this.getRocketMQListener());
+        } else {
+            target = AopProxyUtils.ultimateTargetClass(this.getRocketMQReplyListener());
+        }
+
+        // 初始化消息消费者
+        try {
+            Method method = DefaultRocketMQListenerContainer.class.getDeclaredMethod("initRocketMQPushConsumer");
+            method.setAccessible(true);
+            method.invoke(this.container);
+        } catch (NoSuchMethodException | IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException(e);
+        }
+
+        // 初始化消息类型
+        try {
+            Field field = DefaultRocketMQListenerContainer.class.getDeclaredField("messageType");
+            field.setAccessible(true);
+            field.set(this.container, this.type = RocketMQContextHolder.lookupConsumerMessageType(target));
+        } catch (NoSuchFieldException | IllegalAccessException e) {
+            throw new RuntimeException(e);
+        }
+
+        // 初始化消息处理方法
+        Class<?> clazz;
+        if (this.type instanceof ParameterizedType && this.getMessageConverter() instanceof SmartMessageConverter) {
+            clazz = (Class<?>) ((ParameterizedType) this.type).getRawType();
+        } else if (this.type instanceof Class) {
+            clazz = (Class<?>) this.type;
+        } else {
+            throw new RuntimeException("parameterType:" + this.type + " of onMessage method is not supported");
+        }
+        try {
+            Field field = DefaultRocketMQListenerContainer.class.getDeclaredField("methodParameter");
+            field.setAccessible(true);
+            field.set(this.container, this.parameter = new MethodParameter(target.getMethod("onMessage", clazz), 0));
+        } catch (NoSuchFieldException | IllegalAccessException | NoSuchMethodException e) {
+            throw new RuntimeException(e);
+        }
+
+        // 初始化会话消息监听器
+        if (this.getRocketMQListener() instanceof SessionRocketMQListener) {
+            DefaultMQPushConsumer consumer = this.getConsumer();
+            SessionRocketMQListener listener = (SessionRocketMQListener) this.getRocketMQListener();
+            RocketMQMessageListener annotation = target.getAnnotation(RocketMQMessageListener.class);
+            ConsumeMode mode = ObjectUtils.ifNull(annotation, RocketMQMessageListener::consumeMode);
+            if (mode == ConsumeMode.ORDERLY) {
+                consumer.setMessageListener((MessageListenerOrderly) (messages, context) -> {
+                    try {
+                        listener.onMessageContexts(messages.stream().map(this::convert).collect(Collectors.toList()));
+                    } catch (Exception e) {
+                        log.warn("consume message failed: {}", e.getMessage(), e);
+                        return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
+                    } finally {
+                        consumer.setConsumeMessageBatchMaxSize(listener.getBatchConsumeSize());
+                    }
+                    return ConsumeOrderlyStatus.SUCCESS;
+                });
+            } else {
+                consumer.setMessageListener((MessageListenerConcurrently) (messages, context) -> {
+                    try {
+                        listener.onMessageContexts(messages.stream().map(this::convert).collect(Collectors.toList()));
+                    } catch (Exception e) {
+                        log.warn("consume message failed: {}", e.getMessage(), e);
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+                    } finally {
+                        consumer.setConsumeMessageBatchMaxSize(listener.getBatchConsumeSize());
+                    }
+                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                });
+            }
+        }
+    }
+
+    /**
+     * 消息转换
+     *
+     * @param message 消息
+     * @return 消息
+     */
+    private Object convert(MessageExt message) {
+        if (Objects.equals(this.type, MessageExt.class)) {
+            return message;
+        }
+        MessageConverter converter = this.getMessageConverter();
+        Message<?> body = MessageBuilder.withPayload(new String(message.getBody(), StandardCharsets.UTF_8)).build();
+        MessageContext<?> context =
+                (MessageContext<?>) Objects.requireNonNull(converter.fromMessage(body, MessageContext.class));
+        Object payload = context.getTarget();
+        if (payload == null || (payload instanceof String && this.type == String.class)) {
+            return context;
+        }
+        if (!(payload instanceof String)) {
+            payload = JacksonContextHolder.serialize(payload);
+        }
+        Message<?> target = MessageBuilder.withPayload(payload).build();
+        if (this.type instanceof Class) {
+            payload = converter.fromMessage(target, (Class<?>) this.type);
+        } else {
+            Class<?> clazz = (Class<?>) ((ParameterizedType) this.type).getRawType();
+            payload = ((SmartMessageConverter) converter).fromMessage(target, clazz, this.parameter);
+        }
+        return MessageContext.builder().target(payload).session(context.getSession()).build();
+    }
+
+    @Override
+    public long getSuspendCurrentQueueTimeMillis() {
+        return this.container.getSuspendCurrentQueueTimeMillis();
+    }
+
+    @Override
+    public void setSuspendCurrentQueueTimeMillis(long suspendCurrentQueueTimeMillis) {
+        this.container.setSuspendCurrentQueueTimeMillis(suspendCurrentQueueTimeMillis);
+    }
+
+    @Override
+    public int getDelayLevelWhenNextConsume() {
+        return this.container.getDelayLevelWhenNextConsume();
+    }
+
+    @Override
+    public void setDelayLevelWhenNextConsume(int delayLevelWhenNextConsume) {
+        this.container.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
+    }
+
+    @Override
+    public String getNameServer() {
+        return this.container.getNameServer();
+    }
+
+    @Override
+    public void setNameServer(String nameServer) {
+        this.container.setNameServer(nameServer);
+    }
+
+    @Override
+    public AccessChannel getAccessChannel() {
+        return this.container.getAccessChannel();
+    }
+
+    @Override
+    public void setAccessChannel(AccessChannel accessChannel) {
+        this.container.setAccessChannel(accessChannel);
+    }
+
+    @Override
+    public String getConsumerGroup() {
+        return this.container.getConsumerGroup();
+    }
+
+    @Override
+    public void setConsumerGroup(String consumerGroup) {
+        this.container.setConsumerGroup(consumerGroup);
+    }
+
+    @Override
+    public String getTopic() {
+        return this.container.getTopic();
+    }
+
+    @Override
+    public void setTopic(String topic) {
+        this.container.setTopic(topic);
+    }
+
+    @Override
+    public int getConsumeThreadMax() {
+        return this.container.getConsumeThreadMax();
+    }
+
+    @Override
+    public String getCharset() {
+        return this.container.getCharset();
+    }
+
+    @Override
+    public void setCharset(String charset) {
+        this.container.setCharset(charset);
+    }
+
+    @Override
+    public MessageConverter getMessageConverter() {
+        return this.container.getMessageConverter();
+    }
+
+    @Override
+    public DefaultRocketMQListenerContainer setMessageConverter(MessageConverter messageConverter) {
+        this.container.setMessageConverter(messageConverter);
+        return this;
+    }
+
+    @Override
+    public RocketMQListener<?> getRocketMQListener() {
+        return this.container.getRocketMQListener();
+    }
+
+    @Override
+    public void setRocketMQListener(RocketMQListener rocketMQListener) {
+        this.container.setRocketMQListener(rocketMQListener);
+    }
+
+    @Override
+    public RocketMQReplyListener<?, ?> getRocketMQReplyListener() {
+        return this.container.getRocketMQReplyListener();
+    }
+
+    @Override
+    public void setRocketMQReplyListener(RocketMQReplyListener rocketMQReplyListener) {
+        this.container.setRocketMQReplyListener(rocketMQReplyListener);
+    }
+
+    @Override
+    public RocketMQMessageListener getRocketMQMessageListener() {
+        return this.container.getRocketMQMessageListener();
+    }
+
+    @Override
+    public void setRocketMQMessageListener(RocketMQMessageListener anno) {
+        this.container.setRocketMQMessageListener(anno);
+    }
+
+    @Override
+    public ConsumeMode getConsumeMode() {
+        return this.container.getConsumeMode();
+    }
+
+    @Override
+    public SelectorType getSelectorType() {
+        return this.container.getSelectorType();
+    }
+
+    @Override
+    public void setSelectorExpression(String selectorExpression) {
+        this.container.setSelectorExpression(selectorExpression);
+    }
+
+    @Override
+    public String getSelectorExpression() {
+        return this.container.getSelectorExpression();
+    }
+
+    @Override
+    public MessageModel getMessageModel() {
+        return this.container.getMessageModel();
+    }
+
+    @Override
+    public DefaultMQPushConsumer getConsumer() {
+        return this.container.getConsumer();
+    }
+
+    @Override
+    public void setConsumer(DefaultMQPushConsumer consumer) {
+        this.container.setConsumer(consumer);
+    }
+
+    @Override
+    public void destroy() {
+        this.container.destroy();
+    }
+
+    @Override
+    public boolean isAutoStartup() {
+        return this.container.isAutoStartup();
+    }
+
+    @Override
+    public void stop(Runnable callback) {
+        this.container.stop(callback);
+    }
+
+    @Override
+    public void start() {
+        this.container.start();
+    }
+
+    @Override
+    public void stop() {
+        this.container.stop();
+    }
+
+    @Override
+    public boolean isRunning() {
+        return this.container.isRunning();
+    }
+
+    @Override
+    public int getPhase() {
+        return this.container.getPhase();
+    }
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        this.initialize();
+    }
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        this.container.setApplicationContext(applicationContext);
+    }
+
+    @Override
+    public String toString() {
+        return this.container.toString();
+    }
+
+    @Override
+    public void setName(String name) {
+        this.container.setName(name);
+    }
+}

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/context/RocketMQContextHolder.java

@@ -12,16 +12,21 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 import com.chelvc.framework.base.context.ApplicationContextHolder;
 import com.chelvc.framework.base.context.JacksonContextHolder;
+import com.chelvc.framework.base.context.SessionContextHolder;
+import com.chelvc.framework.base.model.Session;
 import com.chelvc.framework.base.util.ObjectUtils;
 import com.chelvc.framework.base.util.StringUtils;
 import com.chelvc.framework.rocketmq.annotation.MessageDelayer;
+import com.chelvc.framework.rocketmq.interceptor.SessionRocketMQListener;
 import com.chelvc.framework.rocketmq.model.Delay;
 import com.chelvc.framework.rocketmq.model.DelayMessage;
+import com.chelvc.framework.rocketmq.model.MessageContext;
 import com.chelvc.framework.rocketmq.producer.RocketMQTransactionChecker;
 import com.chelvc.framework.rocketmq.producer.RocketMQTransactionExecutor;
 import com.chelvc.framework.rocketmq.producer.RocketMQTransactionHandler;
@@ -63,6 +68,7 @@ import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.env.Environment;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.converter.MessageConverter;
+import org.springframework.messaging.support.GenericMessage;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
 import org.springframework.util.Assert;
@@ -72,14 +78,14 @@ import org.springframework.util.CollectionUtils;
  * RocketMQ上下文工具类
  *
  * @author Woody
- * @date 2023/4/5
+ * @date 2021/9/6
  */
 @Slf4j
 @Component
 @RocketMQTransactionListener
 @RequiredArgsConstructor(onConstructor = @__(@Autowired))
-public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
-        ApplicationContextAware, SmartInitializingSingleton {
+public class RocketMQContextHolder implements RocketMQLocalTransactionListener, ApplicationContextAware,
+        SmartInitializingSingleton {
     /**
      * RocketMQ主题消息头标识
      */
@@ -121,6 +127,11 @@ public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
      */
     private static RocketMQTemplate ROCKETMQ_TEMPLATE;
 
+    /**
+     * 消息转换器
+     */
+    private static MessageConverter MESSAGE_CONVERTER;
+
     /**
      * 消息类型/本地事务检测器映射表
      */
@@ -135,40 +146,222 @@ public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
         CLASS_POOL.appendClassPath(new LoaderClassPath(Thread.currentThread().getContextClassLoader()));
     }
 
-    private final RocketMQMessageConverter rocketmqMessageConverter;
+    @Override
+    @SuppressWarnings("rawtypes")
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        // 初始化环境名称
+        PROFILE = applicationContext.getEnvironment().getProperty("spring.profiles.active");
+
+        // 初始化RocketMQ模版实例
+        ROCKETMQ_TEMPLATE = applicationContext.getBean(RocketMQTemplate.class);
+
+        // 初始化消息转换器
+        MESSAGE_CONVERTER = applicationContext.getBean(RocketMQMessageConverter.class).getMessageConverter();
+
+        // 初始化消息类型/本地事务检测实例映射表
+        Collection<RocketMQTransactionHandler> handlers =
+                applicationContext.getBeansOfType(RocketMQTransactionHandler.class).values();
+        List<RocketMQTransactionChecker> checkers = handlers.stream()
+                .filter(handler -> handler instanceof RocketMQTransactionChecker)
+                .map(handler -> (RocketMQTransactionChecker) handler).collect(Collectors.toList());
+        if (!CollectionUtils.isEmpty(checkers)) {
+            log.info("Loading rocketmq transaction checkers: {}", checkers);
+            MESSAGE_CHECKER_MAPPING = Maps.newHashMapWithExpectedSize(checkers.size());
+            checkers.forEach(checker -> {
+                Class<?> clazz = checker.getPayloadClass();
+                Assert.notNull(clazz, "Rocketmq transaction checker payload class unspecified: " + checker);
+                String type = clazz.getName();
+                Assert.isTrue(!MESSAGE_CHECKER_MAPPING.containsKey(type),
+                        "Rocketmq transaction checker payload class already exists: " + checker);
+                MESSAGE_CHECKER_MAPPING.put(type, checker);
+            });
+        }
+
+        // 初始化消息类型/本地事务执行实例映射表
+        List<RocketMQTransactionExecutor> executors = handlers.stream()
+                .filter(handler -> handler instanceof RocketMQTransactionExecutor)
+                .map(handler -> (RocketMQTransactionExecutor) handler).collect(Collectors.toList());
+        if (!CollectionUtils.isEmpty(executors)) {
+            log.info("Loading rocketmq transaction executors: {}", executors);
+            MESSAGE_EXECUTOR_MAPPING = Maps.newHashMapWithExpectedSize(executors.size());
+            executors.forEach(executor -> {
+                Class<?> clazz = executor.getPayloadClass();
+                Assert.notNull(clazz, "Rocketmq transaction executor payload class unspecified: " + executor);
+                String type = clazz.getName();
+                Assert.isTrue(!MESSAGE_EXECUTOR_MAPPING.containsKey(type),
+                        "Rocketmq transaction executor payload class already exists: " + executor);
+                MESSAGE_EXECUTOR_MAPPING.put(type, executor);
+            });
+        }
+    }
+
+    @Override
+    public void afterSingletonsInstantiated() {
+        // 初始化延时消息延时器
+        ApplicationContext context = ApplicationContextHolder.getApplicationContext();
+        Map<String, Object> consumers = context.getBeansWithAnnotation(MessageDelayer.class);
+        consumers.values().forEach(this::registerMessageDelayerContainer);
+    }
+
+    @Override
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    public RocketMQLocalTransactionState checkLocalTransaction(@NonNull Message message) {
+        RocketMQTransactionChecker checker = Objects.requireNonNull(
+                MESSAGE_CHECKER_MAPPING.get(getMessageType(message)), "Rocketmq transaction checker not found"
+        );
+        return checker.check(message, message2payload(message, checker.getPayloadClass()));
+    }
+
+    @Override
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    public RocketMQLocalTransactionState executeLocalTransaction(@NonNull Message message, Object arg) {
+        try {
+            if (arg instanceof Supplier) {
+                return ((Supplier<RocketMQLocalTransactionState>) arg).get();
+            }
+            RocketMQTransactionExecutor executor = Objects.requireNonNull(
+                    MESSAGE_EXECUTOR_MAPPING.get(getMessageType(message)), "Rocketmq transaction executor not found"
+            );
+            return executor.execute(message, message2payload(message, executor.getPayloadClass()), arg);
+        } catch (RuntimeException e) {
+            // 将异常存入上下文
+            EXCEPTION_CONTEXT.set(e);
+            throw e;
+        }
+    }
 
     /**
-     * 获取消息对象类型
+     * 查找消费者消息类型
      *
-     * @param clazz 消费者对象类型
+     * @param target 消费者对象类型
      * @return 消息对象类型
      */
-    private static Class<?> getMessageType(Class<?> clazz) {
-        Type genericInterface = null;
-        while (Objects.nonNull(clazz)) {
-            Type[] interfaces = clazz.getGenericInterfaces();
+    public static Type lookupConsumerMessageType(@NonNull Class<?> target) {
+        Type parameterized = null;
+        while (Objects.nonNull(target)) {
+            Type[] interfaces = target.getGenericInterfaces();
             for (Type type : interfaces) {
-                if (type instanceof ParameterizedType && (
-                        Objects.equals(((ParameterizedType) type).getRawType(), RocketMQListener.class)
-                                || Objects.equals(((ParameterizedType) type).getRawType(), RocketMQReplyListener.class))
-                ) {
-                    genericInterface = type;
-                    break;
+                if (type instanceof ParameterizedType) {
+                    Type raw = ((ParameterizedType) type).getRawType();
+                    if (Objects.equals(raw, SessionRocketMQListener.class)
+                            || Objects.equals(raw, RocketMQListener.class)
+                            || Objects.equals(raw, RocketMQReplyListener.class)) {
+                        parameterized = type;
+                        break;
+                    }
                 }
             }
-            clazz = clazz.getSuperclass();
+            target = target.getSuperclass();
         }
-        if (Objects.isNull(genericInterface)) {
+        if (Objects.isNull(parameterized)) {
             return Object.class;
         }
 
-        Type[] arguments = ((ParameterizedType) genericInterface).getActualTypeArguments();
+        Type[] arguments = ((ParameterizedType) parameterized).getActualTypeArguments();
         if (Objects.nonNull(arguments) && arguments.length > 0) {
-            return (Class<?>) arguments[0];
+            return arguments[0];
         }
         return Object.class;
     }
 
+    /**
+     * 注册消息延时器消费者容器
+     *
+     * @param consumer 消费者实例
+     */
+    private void registerMessageDelayerContainer(Object consumer) {
+        // 获取延时消息主题
+        Class<?> clazz = AopProxyUtils.ultimateTargetClass(consumer);
+        MessageDelayer annotation = clazz.getAnnotation(MessageDelayer.class);
+        String topic = annotation.topic();
+        if (StringUtils.isEmpty(topic)) {
+            RocketMQMessageListener listener = clazz.getAnnotation(RocketMQMessageListener.class);
+            topic = ObjectUtils.ifNull(listener, RocketMQMessageListener::topic);
+        }
+        Assert.isTrue(StringUtils.nonEmpty(topic), "Message delayer topic unassigned: " + clazz.getName());
+        Environment environment = ApplicationContextHolder.getApplicationContext().getEnvironment();
+        topic = environment.resolvePlaceholders(topic);
+        if (!DELAYER_TOPIC_NAME.add(topic)) {
+            return;
+        }
+
+        // 获取延时消息类型
+        Class<?> type = annotation.type();
+        if (type == Object.class) {
+            type = (Class<?>) lookupConsumerMessageType(clazz);
+        }
+        Assert.isTrue(type != Object.class, "Message delayer type unassigned: " + clazz.getName());
+
+        // 注册消息延时器消费者容器
+        try {
+            Class<?> delayer = this.generateMessageDelayerClass(topic, type);
+            ListenerContainerConfiguration configuration =
+                    ApplicationContextHolder.getApplicationContext().getBean(ListenerContainerConfiguration.class);
+            Method method = ListenerContainerConfiguration.class.getDeclaredMethod(
+                    "registerContainer", String.class, Object.class
+            );
+            method.setAccessible(true);
+            method.invoke(configuration, delayer.getName(), delayer.newInstance());
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    /**
+     * 动态生成消息延时器类对象
+     *
+     * @param topic 延时器消息主题
+     * @param type  消息类型
+     * @return 消息延时器类对象
+     * @throws Exception 处理异常
+     */
+    private Class<?> generateMessageDelayerClass(String topic, Class<?> type) throws Exception {
+        // 构建Class及实现接口
+        CtClass delayer = CLASS_POOL.makeClass(
+                String.format("%s.CustomMessageDelayer_%d", DELAYER_PACKAGE, DELAYER_COUNTER.incrementAndGet())
+        );
+        delayer.setInterfaces(new CtClass[]{CLASS_POOL.get(RocketMQListener.class.getName())});
+        delayer.setGenericSignature(new SignatureAttribute.ClassSignature(null, null,
+                new SignatureAttribute.ClassType[]{new SignatureAttribute.ClassType(
+                        RocketMQListener.class.getName(), new SignatureAttribute.TypeArgument[]{
+                        new SignatureAttribute.TypeArgument(
+                                new SignatureAttribute.ClassType(DelayMessage.class.getName())
+                        )})}).encode()
+        );
+
+        // 添加无参构造方法
+        CtConstructor constructor = new CtConstructor(new CtClass[]{}, delayer);
+        constructor.setBody("{}");
+        delayer.addConstructor(constructor);
+
+        // 添加自定义延时消息消费方法实现
+        delayer.addMethod(CtMethod.make(String.format(
+                "public void onMessage(%s message){%s.processing(\"%s\", %s.class, message);}",
+                DelayMessage.class.getName(),
+                RocketMQContextHolder.class.getName(),
+                topic,
+                type.getName()
+        ), delayer));
+
+        // 添加泛型参数接口方法桥接
+        delayer.addMethod(CtMethod.make(String.format(
+                "public void onMessage(Object message){this.onMessage((%s) message);}",
+                DelayMessage.class.getName()
+        ), delayer));
+
+        // 设置@RocketMQMessageListener注解
+        String unique = getDelayerTopic(topic);
+        ClassFile file = delayer.getClassFile();
+        ConstPool constPool = file.getConstPool();
+        AnnotationsAttribute attribute = new AnnotationsAttribute(constPool, AnnotationsAttribute.visibleTag);
+        Annotation annotation = new Annotation(RocketMQMessageListener.class.getName(), constPool);
+        annotation.addMemberValue("topic", new StringMemberValue(unique, constPool));
+        annotation.addMemberValue("consumerGroup", new StringMemberValue(unique, constPool));
+        attribute.addAnnotation(annotation);
+        file.addAttribute(attribute);
+        return delayer.toClass();
+    }
+
     /**
      * 根据当前环境名称获取实际消息主题
      *
@@ -176,7 +369,7 @@ public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
      * @return 消息主题
      */
     public static String getProfileTopic(@NonNull String topic) {
-        return PROFILE + "-" + topic;
+        return Objects.requireNonNull(PROFILE, "Profile has not been initialized") + "-" + topic;
     }
 
     /**
@@ -198,6 +391,15 @@ public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
         return Objects.requireNonNull(ROCKETMQ_TEMPLATE, "Rocketmq template has not been initialized");
     }
 
+    /**
+     * 获取消息转换器实例
+     *
+     * @return 消息转换器实例
+     */
+    public static MessageConverter getMessageConverter() {
+        return Objects.requireNonNull(MESSAGE_CONVERTER, "Message converter has not been initialized");
+    }
+
     /**
      * 将消息内容转换成消息对象
      *
@@ -205,7 +407,30 @@ public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
      * @return 消息对象实例
      */
     public static Message<?> payload2message(@NonNull Object payload) {
-        return MessageBuilder.withPayload(payload).setHeader(ROCKETMQ_TYPE_HEADER, payload.getClass().getName()).build();
+        Object target = JacksonContextHolder.serialize(payload);
+        Session session = SessionContextHolder.getSession(false);
+        MessageContext<?> context = MessageContext.builder().target(target).session(session).build();
+        return MessageBuilder.withPayload(context).setHeader(ROCKETMQ_TYPE_HEADER, payload.getClass().getName()).build();
+    }
+
+    /**
+     * 将消息对象转换成消息体
+     *
+     * @param message 消息对象
+     * @param type    消息体类型
+     * @param <T>     消息体对象类型
+     * @return 消息体实例
+     */
+    @SuppressWarnings("unchecked")
+    public static <T> T message2payload(@NonNull Message<?> message, @NonNull Class<T> type) {
+        MessageConverter converter = getMessageConverter();
+        MessageContext<?> context =
+                (MessageContext<?>) Objects.requireNonNull(converter.fromMessage(message, MessageContext.class));
+        Object payload = context.getTarget();
+        if (payload == null || (payload instanceof String && type == String.class)) {
+            return (T) payload;
+        }
+        return (T) converter.fromMessage(new GenericMessage<>(payload), type);
     }
 
     /**
@@ -343,6 +568,16 @@ public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
         }
     }
 
+    /**
+     * 异步发送消息
+     *
+     * @param topic   消息主题
+     * @param payload 消息内容
+     */
+    public static void asyncSend(@NonNull String topic, @NonNull Object payload) {
+        asyncSend(topic, payload, true);
+    }
+
     /**
      * 异步发送消息
      *
@@ -573,179 +808,19 @@ public class RocketMQContextHolder implements RocketMQLocalTransactionListener,
         }
     }
 
-    @Override
-    @SuppressWarnings("rawtypes")
-    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
-        // 初始化环境名称
-        PROFILE = applicationContext.getEnvironment().getProperty("spring.profiles.active");
-
-        // 初始化RocketMQ模版实例
-        ROCKETMQ_TEMPLATE = applicationContext.getBean(RocketMQTemplate.class);
-
-        // 初始化消息类型/本地事务检测实例映射表
-        Collection<RocketMQTransactionHandler> handlers =
-                applicationContext.getBeansOfType(RocketMQTransactionHandler.class).values();
-        List<RocketMQTransactionChecker> checkers = handlers.stream()
-                .filter(handler -> handler instanceof RocketMQTransactionChecker)
-                .map(handler -> (RocketMQTransactionChecker) handler).collect(Collectors.toList());
-        if (!CollectionUtils.isEmpty(checkers)) {
-            log.info("Loading rocketmq transaction checkers: {}", checkers);
-            MESSAGE_CHECKER_MAPPING = Maps.newHashMapWithExpectedSize(checkers.size());
-            checkers.forEach(checker -> {
-                Class<?> clazz = checker.getPayloadClass();
-                Assert.notNull(clazz, "Rocketmq transaction checker payload class unspecified: " + checker);
-                String type = clazz.getName();
-                Assert.isTrue(!MESSAGE_CHECKER_MAPPING.containsKey(type),
-                        "Rocketmq transaction checker payload class already exists: " + checker);
-                MESSAGE_CHECKER_MAPPING.put(type, checker);
-            });
-        }
-
-        // 初始化消息类型/本地事务执行实例映射表
-        List<RocketMQTransactionExecutor> executors = handlers.stream()
-                .filter(handler -> handler instanceof RocketMQTransactionExecutor)
-                .map(handler -> (RocketMQTransactionExecutor) handler).collect(Collectors.toList());
-        if (!CollectionUtils.isEmpty(executors)) {
-            log.info("Loading rocketmq transaction executors: {}", executors);
-            MESSAGE_EXECUTOR_MAPPING = Maps.newHashMapWithExpectedSize(executors.size());
-            executors.forEach(executor -> {
-                Class<?> clazz = executor.getPayloadClass();
-                Assert.notNull(clazz, "Rocketmq transaction executor payload class unspecified: " + executor);
-                String type = clazz.getName();
-                Assert.isTrue(!MESSAGE_EXECUTOR_MAPPING.containsKey(type),
-                        "Rocketmq transaction executor payload class already exists: " + executor);
-                MESSAGE_EXECUTOR_MAPPING.put(type, executor);
-            });
-        }
-    }
-
-    @Override
-    public void afterSingletonsInstantiated() {
-        // 初始化延时消息延时器
-        ApplicationContext context = ApplicationContextHolder.getApplicationContext();
-        Map<String, Object> consumers = context.getBeansWithAnnotation(MessageDelayer.class);
-        consumers.values().forEach(this::registerMessageDelayerContainer);
-    }
-
-    @Override
-    @SuppressWarnings({"rawtypes", "unchecked"})
-    public RocketMQLocalTransactionState checkLocalTransaction(@NonNull Message message) {
-        RocketMQTransactionChecker checker = MESSAGE_CHECKER_MAPPING.get(getMessageType(message));
-        return checker.check(message, checker.getPayload(message, this.rocketmqMessageConverter.getMessageConverter()));
-    }
-
-    @Override
-    @SuppressWarnings({"rawtypes", "unchecked"})
-    public RocketMQLocalTransactionState executeLocalTransaction(@NonNull Message message, Object arg) {
-        try {
-            if (arg instanceof Supplier) {
-                return ((Supplier<RocketMQLocalTransactionState>) arg).get();
-            }
-            MessageConverter messageConverter = this.rocketmqMessageConverter.getMessageConverter();
-            RocketMQTransactionExecutor executor = MESSAGE_EXECUTOR_MAPPING.get(getMessageType(message));
-            return executor.execute(message, executor.getPayload(message, messageConverter), arg);
-        } catch (RuntimeException e) {
-            // 将异常存入上下文
-            EXCEPTION_CONTEXT.set(e);
-            throw e;
-        }
-    }
-
     /**
-     * 注册消息延时器消费者容器
+     * 消费消息上下文消息
      *
-     * @param consumer 消费者实例
+     * @param context  消息上下文对象
+     * @param consumer 消息消费者
+     * @param <T>      消息类型
      */
-    private void registerMessageDelayerContainer(Object consumer) {
-        // 获取延时消息主题
-        Class<?> clazz = AopProxyUtils.ultimateTargetClass(consumer);
-        MessageDelayer annotation = clazz.getAnnotation(MessageDelayer.class);
-        String topic = annotation.topic();
-        if (StringUtils.isEmpty(topic)) {
-            RocketMQMessageListener listener = clazz.getAnnotation(RocketMQMessageListener.class);
-            topic = ObjectUtils.ifNull(listener, RocketMQMessageListener::topic);
-        }
-        Assert.isTrue(StringUtils.nonEmpty(topic), "Message delayer topic unassigned: " + clazz.getName());
-        Environment environment = ApplicationContextHolder.getApplicationContext().getEnvironment();
-        topic = environment.resolvePlaceholders(topic);
-        if (!DELAYER_TOPIC_NAME.add(topic)) {
-            return;
-        }
-
-        // 获取延时消息类型
-        Class<?> type = annotation.type();
-        if (type == Object.class) {
-            type = getMessageType(clazz);
-        }
-        Assert.isTrue(type != Object.class, "Message delayer type unassigned: " + clazz.getName());
-
-        // 注册消息延时器消费者容器
+    public static <T> void consumeMessageContext(@NonNull MessageContext<T> context, @NonNull Consumer<T> consumer) {
+        SessionContextHolder.setSession(context.getSession());
         try {
-            Class<?> delayer = this.generateMessageDelayerClass(topic, type);
-            ListenerContainerConfiguration configuration =
-                    ApplicationContextHolder.getApplicationContext().getBean(ListenerContainerConfiguration.class);
-            Method method = ListenerContainerConfiguration.class.getDeclaredMethod(
-                    "registerContainer", String.class, Object.class
-            );
-            method.setAccessible(true);
-            method.invoke(configuration, delayer.getName(), delayer.newInstance());
-        } catch (Exception e) {
-            throw new RuntimeException(e);
+            consumer.accept(context.getTarget());
+        } finally {
+            SessionContextHolder.clearSessionContext();
         }
     }
-
-    /**
-     * 动态生成消息延时器类对象
-     *
-     * @param topic 延时器消息主题
-     * @param type  消息类型
-     * @return 消息延时器类对象
-     * @throws Exception 处理异常
-     */
-    private Class<?> generateMessageDelayerClass(String topic, Class<?> type) throws Exception {
-        // 构建Class及实现接口
-        CtClass delayer = CLASS_POOL.makeClass(
-                String.format("%s.CustomMessageDelayer_%d", DELAYER_PACKAGE, DELAYER_COUNTER.incrementAndGet())
-        );
-        delayer.setInterfaces(new CtClass[]{CLASS_POOL.get(RocketMQListener.class.getName())});
-        delayer.setGenericSignature(new SignatureAttribute.ClassSignature(null, null,
-                new SignatureAttribute.ClassType[]{new SignatureAttribute.ClassType(
-                        RocketMQListener.class.getName(), new SignatureAttribute.TypeArgument[]{
-                        new SignatureAttribute.TypeArgument(
-                                new SignatureAttribute.ClassType(DelayMessage.class.getName())
-                        )})}).encode()
-        );
-
-        // 添加无参构造方法
-        CtConstructor constructor = new CtConstructor(new CtClass[]{}, delayer);
-        constructor.setBody("{}");
-        delayer.addConstructor(constructor);
-
-        // 添加自定义延时消息消费方法实现
-        delayer.addMethod(CtMethod.make(String.format(
-                "public void onMessage(%s message){%s.processing(\"%s\", %s.class, message);}",
-                DelayMessage.class.getName(),
-                RocketMQContextHolder.class.getName(),
-                topic,
-                type.getName()
-        ), delayer));
-
-        // 添加泛型参数接口方法桥接
-        delayer.addMethod(CtMethod.make(String.format(
-                "public void onMessage(Object message){this.onMessage((%s) message);}",
-                DelayMessage.class.getName()
-        ), delayer));
-
-        // 设置@RocketMQMessageListener注解
-        String unique = getDelayerTopic(topic);
-        ClassFile file = delayer.getClassFile();
-        ConstPool constPool = file.getConstPool();
-        AnnotationsAttribute attribute = new AnnotationsAttribute(constPool, AnnotationsAttribute.visibleTag);
-        Annotation annotation = new Annotation(RocketMQMessageListener.class.getName(), constPool);
-        annotation.addMemberValue("topic", new StringMemberValue(unique, constPool));
-        annotation.addMemberValue("consumerGroup", new StringMemberValue(unique, constPool));
-        attribute.addAnnotation(annotation);
-        file.addAttribute(attribute);
-        return delayer.toClass();
-    }
 }

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/interceptor/NativeRocketMQListener.java

@@ -1,211 +0,0 @@
-package com.chelvc.framework.rocketmq.interceptor;
-
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.lang.reflect.ParameterizedType;
-import java.lang.reflect.Type;
-import java.nio.charset.StandardCharsets;
-import java.util.List;
-import java.util.Objects;
-
-import com.chelvc.framework.base.context.ApplicationContextHolder;
-import com.chelvc.framework.base.util.ObjectUtils;
-import lombok.NonNull;
-import lombok.extern.slf4j.Slf4j;
-import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
-import org.apache.rocketmq.common.message.MessageExt;
-import org.apache.rocketmq.spring.annotation.ConsumeMode;
-import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
-import org.apache.rocketmq.spring.core.RocketMQListener;
-import org.apache.rocketmq.spring.core.RocketMQPushConsumerLifecycleListener;
-import org.apache.rocketmq.spring.support.RocketMQMessageConverter;
-import org.springframework.aop.framework.AopProxyUtils;
-import org.springframework.core.MethodParameter;
-import org.springframework.messaging.converter.MessageConverter;
-import org.springframework.messaging.converter.SmartMessageConverter;
-import org.springframework.messaging.support.MessageBuilder;
-
-/**
- * 原生消息监听器实现
- *
- * @author Woody
- * @date 2023/5/8
- */
-@Slf4j
-public class NativeRocketMQListener<T> implements RocketMQPushConsumerLifecycleListener {
-    private final Class<?> target = AopProxyUtils.ultimateTargetClass(this);
-    private Type type;
-    private MethodParameter parameter;
-    private MessageConverter converter;
-    private DefaultMQPushConsumer consumer;
-
-    /**
-     * 初始化消息类型
-     */
-    private void initializeMessageType() {
-        Type parameterized = null;
-        Class<?> clazz = this.target;
-        while (Objects.nonNull(clazz)) {
-            for (Type type : clazz.getGenericInterfaces()) {
-                if (type instanceof ParameterizedType
-                        && Objects.equals(((ParameterizedType) type).getRawType(), RocketMQListener.class)) {
-                    parameterized = type;
-                    break;
-                }
-            }
-            clazz = clazz.getSuperclass();
-        }
-        if (Objects.isNull(parameterized)) {
-            this.type = Object.class;
-        } else {
-            Type[] arguments = ((ParameterizedType) parameterized).getActualTypeArguments();
-            this.type = Objects.nonNull(arguments) && arguments.length > 0 ? arguments[0] : Object.class;
-        }
-    }
-
-    /**
-     * 初始化消息处理方法参数
-     */
-    private void initializeMethodParameter() {
-        Class<?> clazz;
-        Type type = Objects.requireNonNull(this.type, "Message type has not been initialized");
-        if (type instanceof ParameterizedType && this.converter instanceof SmartMessageConverter) {
-            clazz = (Class<?>) ((ParameterizedType) type).getRawType();
-        } else if (type instanceof Class) {
-            clazz = (Class<?>) type;
-        } else {
-            throw new RuntimeException("parameterType:" + type + " of onMessage method is not supported");
-        }
-        try {
-            this.parameter = new MethodParameter(this.target.getMethod("onMessage", clazz), 0);
-        } catch (NoSuchMethodException e) {
-            throw new RuntimeException("parameterType:" + type + " of onMessage method is not supported");
-        }
-    }
-
-    /**
-     * 获取消息批量消费数量
-     *
-     * @return 消息数量
-     */
-    protected int getBatchConsumeSize() {
-        return 1;
-    }
-
-    /**
-     * 消息转换
-     *
-     * @param message 消息
-     * @return 消息
-     */
-    @SuppressWarnings("unchecked")
-    protected T convert(@NonNull MessageExt message) {
-        if (Objects.equals(this.type, MessageExt.class)) {
-            return (T) message;
-        }
-
-        String str = new String(message.getBody(), StandardCharsets.UTF_8);
-        if (Objects.equals(this.type, String.class)) {
-            return (T) str;
-        }
-
-        // If msgType not string, use objectMapper change it.
-        try {
-            if (this.type instanceof Class) {
-                //if the messageType has not Generic Parameter
-                return (T) this.converter.fromMessage(MessageBuilder.withPayload(str).build(), (Class<?>) this.type);
-            }
-            //if the messageType has Generic Parameter, then use SmartMessageConverter#fromMessage with third
-            // parameter "conversionHint".
-            //we have validate the MessageConverter is SmartMessageConverter in this#getMethodParameter.
-            return (T) ((SmartMessageConverter) this.converter).fromMessage(
-                    MessageBuilder.withPayload(str).build(),
-                    (Class<?>) ((ParameterizedType) this.type).getRawType(),
-                    this.parameter
-            );
-        } catch (Exception e) {
-            throw new RuntimeException("cannot convert message to " + this.type, e);
-        }
-    }
-
-    /**
-     * 消费消息
-     *
-     * @param message 消息对象
-     */
-    protected void consume(MessageExt message) {
-        Method method = Objects.requireNonNull(this.parameter.getMethod());
-        try {
-            method.invoke(this, this.convert(message));
-        } catch (IllegalAccessException | InvocationTargetException e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    /**
-     * 顺序消费消息
-     *
-     * @param messages 消息列表
-     * @param context  上下文对象
-     * @return 消费状态
-     */
-    protected ConsumeOrderlyStatus consume(List<MessageExt> messages, ConsumeOrderlyContext context) {
-        messages.forEach(this::consume);
-        return ConsumeOrderlyStatus.SUCCESS;
-    }
-
-    /**
-     * 并发消费消息
-     *
-     * @param messages 消息列表
-     * @param context  上下文对象
-     * @return 消费状态
-     */
-    protected ConsumeConcurrentlyStatus consume(List<MessageExt> messages, ConsumeConcurrentlyContext context) {
-        messages.forEach(this::consume);
-        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-    }
-
-    @Override
-    public void prepareStart(DefaultMQPushConsumer consumer) {
-        // 初始化消息类型及消费方法参数
-        this.consumer = consumer;
-        this.consumer.setConsumeMessageBatchMaxSize(this.getBatchConsumeSize());
-        this.converter = ApplicationContextHolder.getBean(RocketMQMessageConverter.class).getMessageConverter();
-        this.initializeMessageType();
-        this.initializeMethodParameter();
-
-        // 重置消息监听器
-        RocketMQMessageListener annotation = this.target.getAnnotation(RocketMQMessageListener.class);
-        ConsumeMode mode = ObjectUtils.ifNull(annotation, RocketMQMessageListener::consumeMode);
-        if (mode == ConsumeMode.ORDERLY) {
-            this.consumer.setMessageListener((MessageListenerOrderly) (messages, context) -> {
-                try {
-                    return this.consume(messages, context);
-                } catch (Exception e) {
-                    log.warn("consume message failed", e);
-                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
-                } finally {
-                    this.consumer.setConsumeMessageBatchMaxSize(this.getBatchConsumeSize());
-                }
-            });
-        } else {
-            this.consumer.setMessageListener((MessageListenerConcurrently) (messages, context) -> {
-                try {
-                    return this.consume(messages, context);
-                } catch (Exception e) {
-                    log.warn("consume message failed", e);
-                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
-                } finally {
-                    this.consumer.setConsumeMessageBatchMaxSize(this.getBatchConsumeSize());
-                }
-            });
-        }
-    }
-}

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/interceptor/RocketMQConsumeInterceptor.java

@@ -1,34 +0,0 @@
-package com.chelvc.framework.rocketmq.interceptor;
-
-import com.chelvc.framework.base.context.SessionContextHolder;
-import org.aspectj.lang.ProceedingJoinPoint;
-import org.aspectj.lang.annotation.Around;
-import org.aspectj.lang.annotation.Aspect;
-import org.springframework.core.annotation.Order;
-import org.springframework.stereotype.Component;
-
-/**
- * RocketMQ消费者拦截器
- *
- * @author Woody
- * @date 2023/4/5
- */
-@Order
-@Aspect
-@Component
-public class RocketMQConsumeInterceptor {
-    /**
-     * 拦截消息处理
-     *
-     * @param point 接口方法拦截点
-     * @throws Throwable 接口调用异常
-     */
-    @Around("@within(org.apache.rocketmq.spring.annotation.RocketMQMessageListener)")
-    public Object consume(ProceedingJoinPoint point) throws Throwable {
-        try {
-            return point.proceed();
-        } finally {
-            SessionContextHolder.clearSessionContext();
-        }
-    }
-}

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/interceptor/SessionRocketMQListener.java

@@ -0,0 +1,36 @@
+package com.chelvc.framework.rocketmq.interceptor;
+
+import java.util.List;
+
+import com.chelvc.framework.rocketmq.context.RocketMQContextHolder;
+import com.chelvc.framework.rocketmq.model.MessageContext;
+import org.apache.rocketmq.spring.core.RocketMQListener;
+import org.springframework.util.CollectionUtils;
+
+/**
+ * 会话消息监听器接口
+ *
+ * @author Woody
+ * @date 2023/5/18
+ */
+public interface SessionRocketMQListener<T> extends RocketMQListener<T> {
+    /**
+     * 获取消息批量消费数量
+     *
+     * @return 消息数量
+     */
+    default int getBatchConsumeSize() {
+        return 1;
+    }
+
+    /**
+     * 批量处理消息上下文信息
+     *
+     * @param contexts 消息上下文对象列表
+     */
+    default void onMessageContexts(List<MessageContext<T>> contexts) {
+        if (!CollectionUtils.isEmpty(contexts)) {
+            contexts.forEach(context -> RocketMQContextHolder.consumeMessageContext(context, this::onMessage));
+        }
+    }
+}

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/model/MessageContext.java

@@ -0,0 +1,35 @@
+package com.chelvc.framework.rocketmq.model;
+
+import java.io.Serializable;
+
+import com.chelvc.framework.base.model.Session;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.EqualsAndHashCode;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+/**
+ * 消息上下文模型
+ *
+ * @author Woody
+ * @date 2023/5/18
+ */
+@Getter
+@Builder
+@ToString
+@EqualsAndHashCode
+@NoArgsConstructor
+@AllArgsConstructor
+public class MessageContext<T> implements Serializable {
+    /**
+     * 目标对象
+     */
+    private T target;
+
+    /**
+     * 用户会话
+     */
+    private Session session;
+}

framework-rocketmq/src/main/java/com/chelvc/framework/rocketmq/producer/RocketMQTransactionHandler.java

@@ -1,9 +1,5 @@
 package com.chelvc.framework.rocketmq.producer;
 
-import lombok.NonNull;
-import org.springframework.messaging.Message;
-import org.springframework.messaging.converter.MessageConverter;
-
 /**
  * RocketMQ事务处理接口
  *
@@ -18,16 +14,4 @@ public interface RocketMQTransactionHandler<T> {
      * @return 消息内容对象
      */
     Class<T> getPayloadClass();
-
-    /**
-     * 获取消息内容
-     *
-     * @param message   消息对象实例
-     * @param converter 消息转换器实例
-     * @return 消息内容对象实例
-     */
-    @SuppressWarnings("unchecked")
-    default T getPayload(@NonNull Message<?> message, @NonNull MessageConverter converter) {
-        return (T) converter.fromMessage(message, this.getPayloadClass());
-    }
 }

framework-security/src/main/java/com/chelvc/framework/security/context/SecurityContextHolder.java

@@ -17,6 +17,7 @@ import com.chelvc.framework.base.context.ApplicationContextHolder;
 import com.chelvc.framework.base.context.JacksonContextHolder;
 import com.chelvc.framework.base.context.SessionContextHolder;
 import com.chelvc.framework.base.model.Session;
+import com.chelvc.framework.base.model.Terminal;
 import com.chelvc.framework.base.util.ObjectUtils;
 import com.chelvc.framework.base.util.StringUtils;
 import com.chelvc.framework.security.config.SecurityProperties;
@@ -337,7 +338,18 @@ public class SecurityContextHolder implements ApplicationContextAware, ServletRe
      * @return 令牌标识
      */
     public static String buildAccessTokenKey(@NonNull Serializable id) {
-        return String.format("token:%s:%s", id, SessionContextHolder.getTerminal(false));
+        return buildAccessTokenKey(id, SessionContextHolder.getTerminal());
+    }
+
+    /**
+     * 构建令牌标识
+     *
+     * @param id       会话ID
+     * @param terminal 终端信息
+     * @return 令牌标识
+     */
+    public static String buildAccessTokenKey(@NonNull Serializable id, @NonNull Terminal terminal) {
+        return String.format("token:%s:%s", id, terminal);
     }
 
     /**

framework-security/src/main/java/com/chelvc/framework/security/interceptor/TokenArgumentResolver.java

@@ -1,38 +0,0 @@
-package com.chelvc.framework.security.interceptor;
-
-import java.util.List;
-
-import com.chelvc.framework.security.context.SecurityContextHolder;
-import org.springframework.core.MethodParameter;
-import org.springframework.security.oauth2.jwt.Jwt;
-import org.springframework.stereotype.Component;
-import org.springframework.web.bind.support.WebDataBinderFactory;
-import org.springframework.web.context.request.NativeWebRequest;
-import org.springframework.web.method.support.HandlerMethodArgumentResolver;
-import org.springframework.web.method.support.ModelAndViewContainer;
-import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
-
-/**
- * 令牌对象参数转换器
- *
- * @author Woody
- * @date 2023/4/5
- */
-@Component
-public class TokenArgumentResolver implements HandlerMethodArgumentResolver, WebMvcConfigurer {
-    @Override
-    public boolean supportsParameter(MethodParameter parameter) {
-        return Jwt.class.isAssignableFrom(parameter.getParameterType());
-    }
-
-    @Override
-    public Object resolveArgument(MethodParameter parameter, ModelAndViewContainer container,
-                                  NativeWebRequest request, WebDataBinderFactory binderFactory) {
-        return SecurityContextHolder.getJwt();
-    }
-
-    @Override
-    public void addArgumentResolvers(List<HandlerMethodArgumentResolver> resolvers) {
-        resolvers.add(this);
-    }
-}