init

2026-08-13 05:12:07 +08:00 · 2024-03-05 11:39:57 +08:00
parent c9553fdd65
commit f6d1d354e6
1984 changed files with 175140 additions and 2 deletions
@@ -0,0 +1,144 @@
+package com.cf.imes.framework.organ.config;
+
+import com.cf.imes.framework.common.enums.WebFilterOrderEnum;
+import com.cf.imes.framework.mybatis.core.util.MyBatisUtils;
+import com.cf.imes.framework.redis.config.ChenfengCacheProperties;
+import com.cf.imes.framework.organ.core.aop.OrganIgnoreAspect;
+import com.cf.imes.framework.organ.core.db.OrganDatabaseInterceptor;
+import com.cf.imes.framework.organ.core.job.OrganJobAspect;
+import com.cf.imes.framework.organ.core.mq.rabbitmq.OrganRabbitMQInitializer;
+import com.cf.imes.framework.organ.core.mq.redis.OrganRedisMessageInterceptor;
+import com.cf.imes.framework.organ.core.mq.rocketmq.OrganRocketMQInitializer;
+import com.cf.imes.framework.organ.core.redis.OrganRedisCacheManager;
+import com.cf.imes.framework.organ.core.security.OrganSecurityWebFilter;
+import com.cf.imes.framework.organ.core.service.OrganFrameworkService;
+import com.cf.imes.framework.organ.core.service.OrganFrameworkServiceImpl;
+import com.cf.imes.framework.organ.core.web.OrganContextWebFilter;
+import com.cf.imes.framework.web.config.WebProperties;
+import com.cf.imes.framework.web.core.handler.GlobalExceptionHandler;
+import com.cf.imes.module.system.api.organ.OrganApi;
+import com.baomidou.mybatisplus.extension.plugins.MybatisPlusInterceptor;
+import com.baomidou.mybatisplus.extension.plugins.inner.TenantLineInnerInterceptor;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.boot.web.servlet.FilterRegistrationBean;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
+import org.springframework.data.redis.cache.BatchStrategies;
+import org.springframework.data.redis.cache.RedisCacheConfiguration;
+import org.springframework.data.redis.cache.RedisCacheManager;
+import org.springframework.data.redis.cache.RedisCacheWriter;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.core.RedisTemplate;
+
+import java.util.Objects;
+
+@AutoConfiguration
+@ConditionalOnProperty(prefix = "chenfeng.organ", value = "enable", matchIfMissing = true) // 允许使用 chenfeng.organ.enable=false 禁用多组织
+@EnableConfigurationProperties(OrganProperties.class)
+public class ChenfengOrganAutoConfiguration {
+
+    @Bean
+    public OrganFrameworkService tenantFrameworkService(OrganApi organApi) {
+        return new OrganFrameworkServiceImpl(organApi);
+    }
+
+    // ========== AOP ==========
+
+    @Bean
+    public OrganIgnoreAspect organIgnoreAspect() {
+        return new OrganIgnoreAspect();
+    }
+
+    // ========== DB ==========
+
+    @Bean
+    public TenantLineInnerInterceptor tenantLineInnerInterceptor(OrganProperties properties,
+                                                                 MybatisPlusInterceptor interceptor) {
+        TenantLineInnerInterceptor inner = new TenantLineInnerInterceptor(new OrganDatabaseInterceptor(properties));
+        // 添加到 interceptor 中
+        // 需要加在首个，主要是为了在分页插件前面。这个是 MyBatis Plus 的规定
+        MyBatisUtils.addInterceptor(interceptor, inner, 0);
+        return inner;
+    }
+
+    // ========== WEB ==========
+
+    @Bean
+    public FilterRegistrationBean<OrganContextWebFilter> organContextWebFilter() {
+        FilterRegistrationBean<OrganContextWebFilter> registrationBean = new FilterRegistrationBean<>();
+        registrationBean.setFilter(new OrganContextWebFilter());
+        registrationBean.setOrder(WebFilterOrderEnum.TENANT_CONTEXT_FILTER);
+        return registrationBean;
+    }
+
+    // ========== Security ==========
+
+    @Bean
+    public FilterRegistrationBean<OrganSecurityWebFilter> organSecurityWebFilter(OrganProperties tenantProperties,
+                                                                                  WebProperties webProperties,
+                                                                                  GlobalExceptionHandler globalExceptionHandler,
+                                                                                  OrganFrameworkService organFrameworkService) {
+        FilterRegistrationBean<OrganSecurityWebFilter> registrationBean = new FilterRegistrationBean<>();
+        registrationBean.setFilter(new OrganSecurityWebFilter(tenantProperties, webProperties,
+                globalExceptionHandler, organFrameworkService));
+        registrationBean.setOrder(WebFilterOrderEnum.TENANT_SECURITY_FILTER);
+        return registrationBean;
+    }
+
+    // ========== Job ==========
+
+    @Bean
+    @ConditionalOnClass(name = "com.xxl.job.core.handler.annotation.XxlJob")
+    public OrganJobAspect organJobAspect(OrganFrameworkService organFrameworkService) {
+        return new OrganJobAspect(organFrameworkService);
+    }
+
+    // ========== MQ ==========
+
+    /**
+     * 多组织 Redis 消息队列的配置类
+     *
+     * 为什么要单独一个配置类呢？如果直接把 TenantRedisMessageInterceptor Bean 的初始化放外面，会报 RedisMessageInterceptor 类不存在的错误
+     */
+    @Configuration
+    @ConditionalOnClass(name = "com.cf.imes.framework.mq.redis.core.RedisMQTemplate")
+    public static class OrganRedisMQAutoConfiguration {
+
+        @Bean
+        public OrganRedisMessageInterceptor organRedisMessageInterceptor() {
+            return new OrganRedisMessageInterceptor();
+        }
+
+    }
+
+    @Bean
+    @ConditionalOnClass(name = "org.springframework.amqp.rabbit.core.RabbitTemplate")
+    public OrganRabbitMQInitializer organRabbitMQInitializer() {
+        return new OrganRabbitMQInitializer();
+    }
+
+    @Bean
+    @ConditionalOnClass(name = "org.apache.rocketmq.spring.core.RocketMQTemplate")
+    public OrganRocketMQInitializer organRocketMQInitializer() {
+        return new OrganRocketMQInitializer();
+    }
+
+    // ========== Redis ==========
+
+    @Bean
+    @Primary // 引入组织时，tenantRedisCacheManager 为主 Bean
+    public RedisCacheManager organRedisCacheManager(RedisTemplate<String, Object> redisTemplate,
+                                                     RedisCacheConfiguration redisCacheConfiguration,
+                                                     ChenfengCacheProperties chenfengCacheProperties) {
+        // 创建 RedisCacheWriter 对象
+        RedisConnectionFactory connectionFactory = Objects.requireNonNull(redisTemplate.getConnectionFactory());
+        RedisCacheWriter cacheWriter = RedisCacheWriter.nonLockingRedisCacheWriter(connectionFactory,
+                BatchStrategies.scan(chenfengCacheProperties.getRedisScanBatchSize()));
+        // 创建 TenantRedisCacheManager 对象
+        return new OrganRedisCacheManager(cacheWriter, redisCacheConfiguration);
+    }
+}
@@ -0,0 +1,20 @@
+package com.cf.imes.framework.organ.config;
+
+import com.cf.imes.framework.organ.core.rpc.OrganRequestInterceptor;
+import com.cf.imes.module.system.api.organ.OrganApi;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.context.annotation.Bean;
+
+@AutoConfiguration
+@ConditionalOnProperty(prefix = "chenfeng.organ", value = "enable", matchIfMissing = true) // 允许使用 chenfeng.organ.enable=false 禁用多组织
+@EnableFeignClients(clients = OrganApi.class) // 主要是引入相关的 API 服务
+public class ChenfengOrganRpcAutoConfiguration {
+
+    @Bean
+    public OrganRequestInterceptor organRequestInterceptor() {
+        return new OrganRequestInterceptor();
+    }
+
+}
@@ -0,0 +1,42 @@
+package com.cf.imes.framework.organ.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+import java.util.Collections;
+import java.util.Set;
+
+/**
+ * 多组织配置
+ *
+ * @author 晨丰科技
+ */
+@ConfigurationProperties(prefix = "chenfeng.organ")
+@Data
+public class OrganProperties {
+
+    /**
+     * 组织是否开启
+     */
+    private static final Boolean ENABLE_DEFAULT = true;
+
+    /**
+     * 是否开启
+     */
+    private Boolean enable = ENABLE_DEFAULT;
+
+    /**
+     * 需要忽略多组织的请求
+     *
+     * 默认情况下，每个请求需要带上 organ-id 的请求头。但是，部分请求是无需带上的，例如说短信回调、支付回调等 Open API！
+     */
+    private Set<String> ignoreUrls = Collections.emptySet();
+
+    /**
+     * 需要忽略多组织的表
+     *
+     * 即默认所有表都开启多组织的功能，所以记得添加对应的 organ_id 字段哟
+     */
+    private Set<String> ignoreTables = Collections.emptySet();
+
+}
@@ -0,0 +1,18 @@
+package com.cf.imes.framework.organ.core.aop;
+
+import java.lang.annotation.*;
+
+/**
+ * 忽略组织，标记指定方法不进行组织的自动过滤
+ *
+ * 注意，只有 DB 的场景会过滤，其它场景暂时不过滤：
+ * 1、Redis 场景：因为是基于 Key 实现多组织的能力，所以忽略没有意义，不像 DB 是一个 column 实现的
+ * 2、MQ 场景：有点难以抉择，目前可以通过 Consumer 手动在消费的方法上，添加 @OrganIgnore 进行忽略
+ *
+ * @author 晨丰科技
+ */
+@Target({ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+@Inherited
+public @interface OrganIgnore {
+}
@@ -0,0 +1,35 @@
+package com.cf.imes.framework.organ.core.aop;
+
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import com.cf.imes.framework.organ.core.util.OrganUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+
+/**
+ * 忽略多组织的 Aspect，基于 {@link OrganIgnore} 注解实现，用于一些全局的逻辑。
+ * 例如说，一个定时任务，读取所有数据，进行处理。
+ * 又例如说，读取所有数据，进行缓存。
+ *
+ * 整体逻辑的实现，和 {@link OrganUtils#executeIgnore(Runnable)} 需要保持一致
+ *
+ * @author 晨丰科技
+ */
+@Aspect
+@Slf4j
+public class OrganIgnoreAspect {
+
+    @Around("@annotation(organIgnore)")
+    public Object around(ProceedingJoinPoint joinPoint, OrganIgnore organIgnore) throws Throwable {
+        Boolean oldIgnore = OrganContextHolder.isIgnore();
+        try {
+            OrganContextHolder.setIgnore(true);
+            // 执行逻辑
+            return joinPoint.proceed();
+        } finally {
+            OrganContextHolder.setIgnore(oldIgnore);
+        }
+    }
+
+}
@@ -0,0 +1,79 @@
+package com.cf.imes.framework.organ.core.context;
+
+import cn.hutool.core.util.StrUtil;
+import com.cf.imes.framework.common.enums.DocumentEnum;
+import com.alibaba.ttl.TransmittableThreadLocal;
+
+/**
+ * 多组织上下文 Holder
+ *
+ * @author 晨丰科技
+ */
+public class OrganContextHolder {
+
+    /**
+     * 当前组织编号
+     */
+    private static final ThreadLocal<Long> ORGAN_ID = new TransmittableThreadLocal<>();
+
+    /**
+     * 是否忽略组织
+     */
+    private static final ThreadLocal<Boolean> IGNORE = new TransmittableThreadLocal<>();
+
+    /**
+     * 获得组织编号
+     *
+     * @return 组织编号
+     */
+    public static Long getOrganId() {
+        return ORGAN_ID.get();
+    }
+
+    /**
+     * 获得组织编号 String
+     *
+     * @return 组织编号
+     */
+    public static String getOrganIdStr() {
+        Long organId = getOrganId();
+        return StrUtil.toStringOrNull(organId);
+    }
+
+    /**
+     * 获得组织编号。如果不存在，则抛出 NullPointerException 异常
+     *
+     * @return 组织编号
+     */
+    public static Long getRequiredOrganId() {
+        Long organId = getOrganId();
+        if (organId == null) {
+            throw new NullPointerException("OrganContextHolder 不存在组织编号！可参考文档："
+                + DocumentEnum.ORGAN.getUrl());
+        }
+        return organId;
+    }
+
+    public static void setOrganId(Long organId) {
+        ORGAN_ID.set(organId);
+    }
+
+    public static void setIgnore(Boolean ignore) {
+        IGNORE.set(ignore);
+    }
+
+    /**
+     * 当前是否忽略组织
+     *
+     * @return 是否忽略
+     */
+    public static boolean isIgnore() {
+        return Boolean.TRUE.equals(IGNORE.get());
+    }
+
+    public static void clear() {
+        ORGAN_ID.remove();
+        IGNORE.remove();
+    }
+
+}
@@ -0,0 +1,21 @@
+package com.cf.imes.framework.organ.core.db;
+
+import com.cf.imes.framework.mybatis.core.dataobject.BaseDO;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+/**
+ * 拓展多组织的 BaseDO 基类
+ *
+ * @author 晨丰科技
+ */
+@Data
+@EqualsAndHashCode(callSuper = true)
+public abstract class OrganBaseDO extends BaseDO {
+
+    /**
+     * 多组织编号
+     */
+    private Long organId;
+
+}
@@ -0,0 +1,63 @@
+package com.cf.imes.framework.organ.core.db;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.StrUtil;
+import com.cf.imes.framework.organ.config.OrganProperties;
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import com.baomidou.mybatisplus.extension.plugins.handler.TenantLineHandler;
+import com.cf.imes.framework.security.core.LoginUser;
+import com.cf.imes.framework.security.core.util.SecurityFrameworkUtils;
+import com.cf.imes.module.system.enums.permission.RoleCodeEnum;
+import net.sf.jsqlparser.expression.Expression;
+import net.sf.jsqlparser.expression.LongValue;
+
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * 基于 MyBatis Plus 多组织的功能，实现 DB 层面的多组织的功能
+ *
+ * @author 晨丰科技
+ */
+public class OrganDatabaseInterceptor implements TenantLineHandler {
+
+    private final Set<String> ignoreTables = new HashSet<>();
+
+    public OrganDatabaseInterceptor(OrganProperties properties) {
+        // 不同 DB 下，大小写的习惯不同，所以需要都添加进去
+        properties.getIgnoreTables().forEach(table -> {
+            ignoreTables.add(table.toLowerCase());
+            ignoreTables.add(table.toUpperCase());
+        });
+        // 在 OracleKeyGenerator 中，生成主键时，会查询这个表，查询这个表后，会自动拼接 ORGAN_ID 导致报错
+        ignoreTables.add("DUAL");
+    }
+
+    @Override
+    public Expression getTenantId() {
+        return new LongValue(OrganContextHolder.getRequiredOrganId());
+    }
+
+    @Override
+    public String getTenantIdColumn() {
+        return "organ_id";
+    }
+
+    @Override
+    public boolean ignoreTable(String tableName) {
+        return OrganContextHolder.isIgnore() // 情况一，全局忽略多组织
+            || CollUtil.contains(ignoreTables, tableName) // 情况二，忽略多组织的表
+            || isAdmin(); //情况三，忽略超级管理员
+    }
+    private boolean isAdmin() {
+        LoginUser loginUser = SecurityFrameworkUtils.getLoginUser();
+        if(!Objects.isNull(loginUser)) {
+            Boolean isSupAdmin = loginUser.getIsSupAdmin();
+            return Objects.isNull(isSupAdmin)? Boolean.FALSE : isSupAdmin;
+        }
+        return Boolean.FALSE;
+    }
+
+}
@@ -0,0 +1,14 @@
+package com.cf.imes.framework.organ.core.job;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+/**
+ * 多组织 Job 注解
+ */
+@Target({ElementType.METHOD})
+@Retention(RetentionPolicy.RUNTIME)
+public @interface OrganJob {
+}
@@ -0,0 +1,65 @@
+package com.cf.imes.framework.organ.core.job;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.exceptions.ExceptionUtil;
+import cn.hutool.core.util.StrUtil;
+import com.cf.imes.framework.common.util.json.JsonUtils;
+import com.cf.imes.framework.organ.core.service.OrganFrameworkService;
+import com.cf.imes.framework.organ.core.util.OrganUtils;
+import com.xxl.job.core.context.XxlJobHelper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 多组织 JobHandler AOP
+ * 任务执行时，会按照组织逐个执行 Job 的逻辑
+ *
+ * 注意，需要保证 JobHandler 的幂等性。因为 Job 因为某个组织执行失败重试时，之前执行成功的组织也会再次执行。
+ *
+ * @author 晨丰科技
+ */
+@Aspect
+@RequiredArgsConstructor
+@Slf4j
+public class OrganJobAspect {
+
+    private final OrganFrameworkService organFrameworkService;
+
+    @Around("@annotation(organJob)")
+    public void around(ProceedingJoinPoint joinPoint, OrganJob organJob) {
+        // 获得组织列表
+        List<Long> organIds = organFrameworkService.getOrganIds();
+        if (CollUtil.isEmpty(organIds)) {
+            return;
+        }
+
+        // 逐个组织，执行 Job
+        Map<Long, String> results = new ConcurrentHashMap<>();
+        organIds.parallelStream().forEach(organId -> {
+            // TODO 晨丰：先通过 parallel 实现并行；1）多个组织，是一条执行日志；2）异常的情况
+            OrganUtils.execute(organId, () -> {
+                try {
+                    joinPoint.proceed();
+                } catch (Throwable e) {
+                    results.put(organId, ExceptionUtil.getRootCauseMessage(e));
+                    // 打印异常
+                    XxlJobHelper.log(StrUtil.format("[多组织({}) 执行任务({})，发生异常：{}]",
+                            organId, joinPoint.getSignature(), ExceptionUtils.getStackTrace(e)));
+                }
+            });
+        });
+        // 如果 results 非空，说明发生了异常，标记 XXL-Job 执行失败
+        if (CollUtil.isNotEmpty(results)) {
+            XxlJobHelper.handleFail(JsonUtils.toJsonString(results));
+        }
+    }
+
+}
@@ -0,0 +1,37 @@
+package com.cf.imes.framework.organ.core.mq.kafka;
+
+import cn.hutool.core.util.StrUtil;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.env.EnvironmentPostProcessor;
+import org.springframework.core.env.ConfigurableEnvironment;
+
+/**
+ * 多组织的 Kafka 的 {@link EnvironmentPostProcessor} 实现类
+ *
+ * Kafka Producer 发送消息时，增加 {@link OrganKafkaProducerInterceptor} 拦截器
+ *
+ * @author 晨丰科技
+ */
+@Slf4j
+public class OrganKafkaEnvironmentPostProcessor implements EnvironmentPostProcessor {
+
+    private static final String PROPERTY_KEY_INTERCEPTOR_CLASSES = "spring.kafka.producer.properties.interceptor.classes";
+
+    @Override
+    public void postProcessEnvironment(ConfigurableEnvironment environment, SpringApplication application) {
+        // 添加 OrganKafkaProducerInterceptor 拦截器
+        try {
+            String value = environment.getProperty(PROPERTY_KEY_INTERCEPTOR_CLASSES);
+            if (StrUtil.isEmpty(value)) {
+                value = OrganKafkaProducerInterceptor.class.getName();
+            } else {
+                value += "," + OrganKafkaProducerInterceptor.class.getName();
+            }
+            environment.getSystemProperties().put(PROPERTY_KEY_INTERCEPTOR_CLASSES, value);
+        } catch (NoClassDefFoundError ignore) {
+            // 如果触发 NoClassDefFoundError 异常，说明 TenantKafkaProducerInterceptor 类不存在，即没引入 kafka-spring 依赖
+        }
+    }
+
+}
@@ -0,0 +1,47 @@
+package com.cf.imes.framework.organ.core.mq.kafka;
+
+import cn.hutool.core.util.ReflectUtil;
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import org.apache.kafka.clients.producer.ProducerInterceptor;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.apache.kafka.common.header.Headers;
+import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+
+import java.util.Map;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * Kafka 消息队列的多组织 {@link ProducerInterceptor} 实现类
+ *
+ * 1. Producer 发送消息时，将 {@link OrganContextHolder} 组织编号，添加到消息的 Header 中
+ * 2. Consumer 消费消息时，将消息的 Header 的组织编号，添加到 {@link OrganContextHolder} 中，通过 {@link InvocableHandlerMethod} 实现
+ *
+ * @author 晨丰科技
+ */
+public class OrganKafkaProducerInterceptor implements ProducerInterceptor<Object, Object> {
+
+    @Override
+    public ProducerRecord<Object, Object> onSend(ProducerRecord<Object, Object> record) {
+        Long organId = OrganContextHolder.getOrganId();
+        if (organId != null) {
+            Headers headers = (Headers) ReflectUtil.getFieldValue(record, "headers"); // private 属性，没有 get 方法，智能反射
+            headers.add(HEADER_ORGAN_ID, organId.toString().getBytes());
+        }
+        return record;
+    }
+
+    @Override
+    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
+    }
+
+    @Override
+    public void close() {
+    }
+
+    @Override
+    public void configure(Map<String, ?> configs) {
+    }
+
+}
@@ -0,0 +1,23 @@
+package com.cf.imes.framework.organ.core.mq.rabbitmq;
+
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+
+/**
+ * 多组织的 RabbitMQ 初始化器
+ *
+ * @author 晨丰科技
+ */
+public class OrganRabbitMQInitializer implements BeanPostProcessor {
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        if (bean instanceof RabbitTemplate) {
+            RabbitTemplate rabbitTemplate = (RabbitTemplate) bean;
+            rabbitTemplate.addBeforePublishPostProcessors(new OrganRabbitMQMessagePostProcessor());
+        }
+        return bean;
+    }
+
+}
@@ -0,0 +1,31 @@
+package com.cf.imes.framework.organ.core.mq.rabbitmq;
+
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import org.apache.kafka.clients.producer.ProducerInterceptor;
+import org.springframework.amqp.AmqpException;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessagePostProcessor;
+import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * RabbitMQ 消息队列的多组织 {@link ProducerInterceptor} 实现类
+ *
+ * 1. Producer 发送消息时，将 {@link OrganContextHolder} 组织编号，添加到消息的 Header 中
+ * 2. Consumer 消费消息时，将消息的 Header 的组织编号，添加到 {@link OrganContextHolder} 中，通过 {@link InvocableHandlerMethod} 实现
+ *
+ * @author 晨丰科技
+ */
+public class OrganRabbitMQMessagePostProcessor implements MessagePostProcessor {
+
+    @Override
+    public Message postProcessMessage(Message message) throws AmqpException {
+        Long organId = OrganContextHolder.getOrganId();
+        if (organId != null) {
+            message.getMessageProperties().getHeaders().put(HEADER_ORGAN_ID, organId);
+        }
+        return message;
+    }
+
+}
@@ -0,0 +1,42 @@
+package com.cf.imes.framework.organ.core.mq.redis;
+
+import cn.hutool.core.util.StrUtil;
+import com.cf.imes.framework.mq.redis.core.interceptor.RedisMessageInterceptor;
+import com.cf.imes.framework.mq.redis.core.message.AbstractRedisMessage;
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * 多组织 {@link AbstractRedisMessage} 拦截器
+ *
+ * 1. Producer 发送消息时，将 {@link OrganContextHolder} 组织编号，添加到消息的 Header 中
+ * 2. Consumer 消费消息时，将消息的 Header 的组织编号，添加到 {@link OrganContextHolder} 中
+ *
+ * @author 晨丰科技
+ */
+public class OrganRedisMessageInterceptor implements RedisMessageInterceptor {
+
+    @Override
+    public void sendMessageBefore(AbstractRedisMessage message) {
+        Long organId = OrganContextHolder.getOrganId();
+        if (organId != null) {
+            message.addHeader(HEADER_ORGAN_ID, organId.toString());
+        }
+    }
+
+    @Override
+    public void consumeMessageBefore(AbstractRedisMessage message) {
+        String organIdStr = message.getHeader(HEADER_ORGAN_ID);
+        if (StrUtil.isNotEmpty(organIdStr)) {
+            OrganContextHolder.setOrganId(Long.valueOf(organIdStr));
+        }
+    }
+
+    @Override
+    public void consumeMessageAfter(AbstractRedisMessage message) {
+        // 注意，Consumer 是一个逻辑的入口，所以不考虑原本上下文就存在组织编号的情况
+        OrganContextHolder.clear();
+    }
+
+}
@@ -0,0 +1,46 @@
+package com.cf.imes.framework.organ.core.mq.rocketmq;
+
+import cn.hutool.core.lang.Assert;
+import cn.hutool.core.util.StrUtil;
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import org.apache.rocketmq.client.hook.ConsumeMessageContext;
+import org.apache.rocketmq.client.hook.ConsumeMessageHook;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.springframework.messaging.handler.invocation.InvocableHandlerMethod;
+
+import java.util.List;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * RocketMQ 消息队列的多组织 {@link ConsumeMessageHook} 实现类
+ *
+ * Consumer 消费消息时，将消息的 Header 的组织编号，添加到 {@link OrganContextHolder} 中，通过 {@link InvocableHandlerMethod} 实现
+ *
+ * @author 晨丰科技
+ */
+public class OrganRocketMQConsumeMessageHook implements ConsumeMessageHook {
+
+    @Override
+    public String hookName() {
+        return getClass().getSimpleName();
+    }
+
+    @Override
+    public void consumeMessageBefore(ConsumeMessageContext context) {
+        // 校验，消息必须是单条，不然设置组织可能不正确
+        List<MessageExt> messages = context.getMsgList();
+        Assert.isTrue(messages.size() == 1, "消息条数({})不正确", messages.size());
+        // 设置组织编号
+        String organId = messages.get(0).getUserProperty(HEADER_ORGAN_ID);
+        if (StrUtil.isNotEmpty(organId)) {
+            OrganContextHolder.setOrganId(Long.parseLong(organId));
+        }
+    }
+
+    @Override
+    public void consumeMessageAfter(ConsumeMessageContext context) {
+        OrganContextHolder.clear();
+    }
+
+}
@@ -0,0 +1,53 @@
+package com.cf.imes.framework.organ.core.mq.rocketmq;
+
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.impl.consumer.DefaultMQPushConsumerImpl;
+import org.apache.rocketmq.client.impl.producer.DefaultMQProducerImpl;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+
+/**
+ * 多组织的 RocketMQ 初始化器
+ *
+ * @author 晨丰科技
+ */
+public class OrganRocketMQInitializer implements BeanPostProcessor {
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        if (bean instanceof DefaultRocketMQListenerContainer) {
+            DefaultRocketMQListenerContainer container = (DefaultRocketMQListenerContainer) bean;
+            initOrganConsumer(container.getConsumer());
+        } else if (bean instanceof RocketMQTemplate) {
+            RocketMQTemplate template = (RocketMQTemplate) bean;
+            initOrganProducer(template.getProducer());
+        }
+        return bean;
+    }
+
+    private void initOrganProducer(DefaultMQProducer producer) {
+        if (producer == null) {
+            return;
+        }
+        DefaultMQProducerImpl producerImpl = producer.getDefaultMQProducerImpl();
+        if (producerImpl == null) {
+            return;
+        }
+        producerImpl.registerSendMessageHook(new OrganRocketMQSendMessageHook());
+    }
+
+    private void initOrganConsumer(DefaultMQPushConsumer consumer) {
+        if (consumer == null) {
+            return;
+        }
+        DefaultMQPushConsumerImpl consumerImpl = consumer.getDefaultMQPushConsumerImpl();
+        if (consumerImpl == null) {
+            return;
+        }
+        consumerImpl.registerConsumeMessageHook(new OrganRocketMQConsumeMessageHook());
+    }
+
+}
@@ -0,0 +1,36 @@
+package com.cf.imes.framework.organ.core.mq.rocketmq;
+
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import org.apache.rocketmq.client.hook.SendMessageContext;
+import org.apache.rocketmq.client.hook.SendMessageHook;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * RocketMQ 消息队列的多组织 {@link SendMessageHook} 实现类
+ *
+ * Producer 发送消息时，将 {@link OrganContextHolder} 组织编号，添加到消息的 Header 中
+ *
+ * @author 晨丰科技
+ */
+public class OrganRocketMQSendMessageHook implements SendMessageHook {
+
+    @Override
+    public String hookName() {
+        return getClass().getSimpleName();
+    }
+
+    @Override
+    public void sendMessageBefore(SendMessageContext sendMessageContext) {
+        Long organId = OrganContextHolder.getOrganId();
+        if (organId == null) {
+            return;
+        }
+        sendMessageContext.getMessage().putUserProperty(HEADER_ORGAN_ID, organId.toString());
+    }
+
+    @Override
+    public void sendMessageAfter(SendMessageContext sendMessageContext) {
+    }
+
+}
@@ -0,0 +1,38 @@
+package com.cf.imes.framework.organ.core.redis;
+
+import com.cf.imes.framework.redis.core.TimeoutRedisCacheManager;
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.cache.Cache;
+import org.springframework.data.redis.cache.RedisCacheConfiguration;
+import org.springframework.data.redis.cache.RedisCacheManager;
+import org.springframework.data.redis.cache.RedisCacheWriter;
+
+/**
+ * 多组织的 {@link RedisCacheManager} 实现类
+ *
+ * 操作指定 name 的 {@link Cache} 时，自动拼接组织后缀，格式为 name + ":" + organId + 后缀
+ *
+ * @author airhead
+ */
+@Slf4j
+public class OrganRedisCacheManager extends TimeoutRedisCacheManager {
+
+    public OrganRedisCacheManager(RedisCacheWriter cacheWriter,
+                                  RedisCacheConfiguration defaultCacheConfiguration) {
+        super(cacheWriter, defaultCacheConfiguration);
+    }
+
+    @Override
+    public Cache getCache(String name) {
+        // 如果开启多组织，则 name 拼接组织后缀
+        if (!OrganContextHolder.isIgnore()
+            && OrganContextHolder.getOrganId() != null) {
+            name = name + ":" + OrganContextHolder.getOrganId();
+        }
+
+        // 继续基于父方法
+        return super.getCache(name);
+    }
+
+}
@@ -0,0 +1,24 @@
+package com.cf.imes.framework.organ.core.rpc;
+
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import feign.RequestInterceptor;
+import feign.RequestTemplate;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * Tenant 的 RequestInterceptor 实现类：Feign 请求时，将 {@link OrganContextHolder} 设置到 header 中，继续透传给被调用的服务
+ *
+ * @author 晨丰科技
+ */
+public class OrganRequestInterceptor implements RequestInterceptor {
+
+    @Override
+    public void apply(RequestTemplate requestTemplate) {
+        Long organId = OrganContextHolder.getOrganId();
+        if (organId != null) {
+            requestTemplate.header(HEADER_ORGAN_ID, String.valueOf(organId));
+        }
+    }
+
+}
@@ -0,0 +1,130 @@
+package com.cf.imes.framework.organ.core.security;
+
+import cn.hutool.core.collection.CollUtil;
+import cn.hutool.core.util.StrUtil;
+import com.cf.imes.framework.common.enums.RpcConstants;
+import com.cf.imes.framework.common.exception.enums.GlobalErrorCodeConstants;
+import com.cf.imes.framework.common.pojo.CommonResult;
+import com.cf.imes.framework.common.util.servlet.ServletUtils;
+import com.cf.imes.framework.security.core.LoginUser;
+import com.cf.imes.framework.security.core.util.SecurityFrameworkUtils;
+import com.cf.imes.framework.organ.config.OrganProperties;
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import com.cf.imes.framework.organ.core.service.OrganFrameworkService;
+import com.cf.imes.framework.web.config.WebProperties;
+import com.cf.imes.framework.web.core.filter.ApiRequestFilter;
+import com.cf.imes.framework.web.core.handler.GlobalExceptionHandler;
+import com.cf.imes.framework.web.core.util.WebFrameworkUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.AntPathMatcher;
+
+import javax.servlet.FilterChain;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+import java.util.Objects;
+
+/**
+ * 多组织 Security Web 过滤器
+ * 1. 如果是登陆的用户，校验是否有权限访问该组织，避免越权问题。
+ * 2. 如果请求未带组织的编号，检查是否是忽略的 URL，否则也不允许访问。
+ * 3. 校验组织是合法，例如说被禁用、到期
+ *
+ * 校验用户访问的组织，是否是其所在的组织，
+ *
+ * @author 晨丰科技
+ */
+@Slf4j
+public class OrganSecurityWebFilter extends ApiRequestFilter {
+
+    private final OrganProperties organProperties;
+
+    private final AntPathMatcher pathMatcher;
+
+    private final GlobalExceptionHandler globalExceptionHandler;
+    private final OrganFrameworkService organFrameworkService;
+
+    public OrganSecurityWebFilter(OrganProperties organProperties,
+                                  WebProperties webProperties,
+                                  GlobalExceptionHandler globalExceptionHandler,
+                                  OrganFrameworkService organFrameworkService) {
+        super(webProperties);
+        this.organProperties = organProperties;
+        this.pathMatcher = new AntPathMatcher();
+        this.globalExceptionHandler = globalExceptionHandler;
+        this.organFrameworkService = organFrameworkService;
+    }
+
+    @Override
+    protected boolean shouldNotFilter(HttpServletRequest request) {
+        return super.shouldNotFilter(request) &&
+                !StrUtil.startWithAny(request.getRequestURI(), RpcConstants.RPC_API_PREFIX); // 因为 RPC API 也会透传组织编号
+    }
+
+    @Override
+    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain chain)
+            throws ServletException, IOException {
+        Long organId = OrganContextHolder.getOrganId();
+        boolean isRpcRequest = WebFrameworkUtils.isRpcRequest(request);
+        // 1. 登陆的用户，校验是否有权限访问该组织，避免越权问题。
+        LoginUser user = SecurityFrameworkUtils.getLoginUser();
+        if (user != null) {
+            // 如果获取不到组织编号，则尝试使用登陆用户的组织编号
+            if (organId == null) {
+                organId = user.getOrganId();
+                OrganContextHolder.setOrganId(organId);
+            // 如果传递了组织编号，则进行比对组织编号，避免越权问题
+            } else if (!Objects.equals(user.getOrganId(), OrganContextHolder.getOrganId())
+                && !isRpcRequest) { // Cloud 特殊逻辑：如果是 RPC 请求，就不校验了。主要考虑，一些场景下，会调用 OrganUtils 去切换组织
+                log.error("[doFilterInternal][组织({}) User({}/{}) 越权访问组织({}) URL({}/{})]",
+                        user.getOrganId(), user.getId(), user.getUserType(),
+                        OrganContextHolder.getOrganId(), request.getRequestURI(), request.getMethod());
+                ServletUtils.writeJSON(response, CommonResult.error(GlobalErrorCodeConstants.FORBIDDEN.getCode(),
+                        "您无权访问该组织的数据"));
+                return;
+            }
+        }
+
+        // 如果非允许忽略组织的 URL，则校验组织是否合法
+        if (!isIgnoreUrl(request)) {
+            // 2. 如果请求未带组织的编号，不允许访问。
+            if (organId == null) {
+                log.error("[doFilterInternal][URL({}/{}) 未传递组织编号]", request.getRequestURI(), request.getMethod());
+                ServletUtils.writeJSON(response, CommonResult.error(GlobalErrorCodeConstants.BAD_REQUEST.getCode(),
+                        "请求的组织标识未传递，请进行排查"));
+                return;
+            }
+            // 3. 校验组织是合法，例如说被禁用、到期
+            try {
+                organFrameworkService.validOrgan(organId);
+            } catch (Throwable ex) {
+                CommonResult<?> result = globalExceptionHandler.allExceptionHandler(request, ex);
+                ServletUtils.writeJSON(response, result);
+                return;
+            }
+        } else { // 如果是允许忽略组织的 URL，若未传递组织编号，则默认忽略组织编号，避免报错
+            if (organId == null) {
+                OrganContextHolder.setIgnore(true);
+            }
+        }
+
+        // 继续过滤
+        chain.doFilter(request, response);
+    }
+
+    private boolean isIgnoreUrl(HttpServletRequest request) {
+        // 快速匹配，保证性能
+        if (CollUtil.contains(organProperties.getIgnoreUrls(), request.getRequestURI())) {
+            return true;
+        }
+        // 逐个 Ant 路径匹配
+        for (String url : organProperties.getIgnoreUrls()) {
+            if (pathMatcher.match(url, request.getRequestURI())) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+}
@@ -0,0 +1,26 @@
+package com.cf.imes.framework.organ.core.service;
+
+import java.util.List;
+
+/**
+ * Organ 框架 Service 接口，定义获取组织信息
+ *
+ * @author 晨丰科技
+ */
+public interface OrganFrameworkService {
+
+    /**
+     * 获得所有组织
+     *
+     * @return 组织编号数组
+     */
+    List<Long> getOrganIds();
+
+    /**
+     * 校验组织是否合法
+     *
+     * @param id 组织编号
+     */
+    void validOrgan(Long id);
+
+}
@@ -0,0 +1,64 @@
+package com.cf.imes.framework.organ.core.service;
+
+import com.cf.imes.framework.common.pojo.CommonResult;
+import com.cf.imes.framework.common.util.cache.CacheUtils;
+import com.cf.imes.module.system.api.organ.OrganApi;
+import com.google.common.cache.CacheLoader;
+import com.google.common.cache.LoadingCache;
+import lombok.RequiredArgsConstructor;
+import lombok.SneakyThrows;
+
+import java.time.Duration;
+import java.util.List;
+
+/**
+ * Organ 框架 Service 实现类
+ *
+ * @author 晨丰科技
+ */
+@RequiredArgsConstructor
+public class OrganFrameworkServiceImpl implements OrganFrameworkService {
+
+    private final OrganApi organApi;
+
+    /**
+     * 针对 {@link #getOrganIds()} 的缓存
+     */
+    private final LoadingCache<Object, List<Long>> getOrganIdsCache = CacheUtils.buildAsyncReloadingCache(
+            Duration.ofMinutes(1L), // 过期时间 1 分钟
+            new CacheLoader<Object, List<Long>>() {
+
+                @Override
+                public List<Long> load(Object key) {
+                    return organApi.getOrganIdList().getCheckedData();
+                }
+
+            });
+
+    /**
+     * 针对 {@link #validOrgan(Long)} 的缓存
+     */
+    private final LoadingCache<Long, CommonResult<Boolean>> validOrganCache = CacheUtils.buildAsyncReloadingCache(
+            Duration.ofMinutes(1L), // 过期时间 1 分钟
+            new CacheLoader<Long, CommonResult<Boolean>>() {
+
+                @Override
+                public CommonResult<Boolean> load(Long id) {
+                    return organApi.validOrgan(id);
+                }
+
+            });
+
+    @Override
+    @SneakyThrows
+    public List<Long> getOrganIds() {
+        return getOrganIdsCache.get(Boolean.TRUE);
+    }
+
+    @Override
+    @SneakyThrows
+    public void validOrgan(Long id) {
+        validOrganCache.get(id).checkError();
+    }
+
+}
@@ -0,0 +1,93 @@
+package com.cf.imes.framework.organ.core.util;
+
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+
+import java.util.Map;
+import java.util.concurrent.Callable;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * 多组织 Util
+ *
+ * @author 晨丰科技
+ */
+public class OrganUtils {
+
+    /**
+     * 使用指定组织，执行对应的逻辑
+     *
+     * 注意，如果当前是忽略组织的情况下，会被强制设置成不忽略组织
+     * 当然，执行完成后，还是会恢复回去
+     *
+     * @param organId 组织编号
+     * @param runnable 逻辑
+     */
+    public static void execute(Long organId, Runnable runnable) {
+        Long oldorganId = OrganContextHolder.getOrganId();
+        Boolean oldIgnore = OrganContextHolder.isIgnore();
+        try {
+            OrganContextHolder.setOrganId(organId);
+            OrganContextHolder.setIgnore(false);
+            // 执行逻辑
+            runnable.run();
+        } finally {
+            OrganContextHolder.setOrganId(oldorganId);
+            OrganContextHolder.setIgnore(oldIgnore);
+        }
+    }
+
+    /**
+     * 使用指定组织，执行对应的逻辑
+     *
+     * 注意，如果当前是忽略组织的情况下，会被强制设置成不忽略组织
+     * 当然，执行完成后，还是会恢复回去
+     *
+     * @param organId 组织编号
+     * @param callable 逻辑
+     */
+    public static <V> V execute(Long organId, Callable<V> callable) {
+        Long oldorganId = OrganContextHolder.getOrganId();
+        Boolean oldIgnore = OrganContextHolder.isIgnore();
+        try {
+            OrganContextHolder.setOrganId(organId);
+            OrganContextHolder.setIgnore(false);
+            // 执行逻辑
+            return callable.call();
+        } catch (Exception e) {
+            throw new RuntimeException(e);
+        } finally {
+            OrganContextHolder.setOrganId(oldorganId);
+            OrganContextHolder.setIgnore(oldIgnore);
+        }
+    }
+
+    /**
+     * 忽略组织，执行对应的逻辑
+     *
+     * @param runnable 逻辑
+     */
+    public static void executeIgnore(Runnable runnable) {
+        Boolean oldIgnore = OrganContextHolder.isIgnore();
+        try {
+            OrganContextHolder.setIgnore(true);
+            // 执行逻辑
+            runnable.run();
+        } finally {
+            OrganContextHolder.setIgnore(oldIgnore);
+        }
+    }
+
+    /**
+     * 将多组织编号，添加到 header 中
+     *
+     * @param headers HTTP 请求 headers
+     * @param organId 组织编号
+     */
+    public static void addTenantHeader(Map<String, String> headers, Long organId) {
+        if (organId != null) {
+            headers.put(HEADER_ORGAN_ID, organId.toString());
+        }
+    }
+
+}
@@ -0,0 +1,37 @@
+package com.cf.imes.framework.organ.core.web;
+
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import com.cf.imes.framework.web.core.util.WebFrameworkUtils;
+import org.springframework.web.filter.OncePerRequestFilter;
+
+import javax.servlet.FilterChain;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+
+/**
+ * 多组织 Context Web 过滤器
+ * 将请求 Header 中的 tenant-id 解析出来，添加到 {@link OrganContextHolder} 中，这样后续的 DB 等操作，可以获得到组织编号。
+ *
+ * @author 晨丰科技
+ */
+public class OrganContextWebFilter extends OncePerRequestFilter {
+
+    @Override
+    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain chain)
+            throws ServletException, IOException {
+        // 设置
+        Long organId = WebFrameworkUtils.getOrganId(request);
+        if (organId != null) {
+            OrganContextHolder.setOrganId(organId);
+        }
+        try {
+            chain.doFilter(request, response);
+        } finally {
+            // 清理
+            OrganContextHolder.clear();
+        }
+    }
+
+}
@@ -0,0 +1,17 @@
+/**
+ * 多组织，支持如下层面：
+ * 1. DB：基于 MyBatis Plus 多组织的功能实现。
+ * 2. Redis：通过在 Redis Key 上拼接组织编号的方式，进行隔离。
+ * 3. Web：请求 HTTP API 时，解析 Header 的 organ-id 组织编号，添加到组织上下文。
+ * 4. Security：校验当前登陆的用户，是否越权访问其它组织的数据。
+ * 5. Job：在 JobHandler 执行任务时，会按照每个组织，都独立并行执行一次。
+ * 6. MQ：在 Producer 发送消息时，Header 带上 organ-id 组织编号；在 Consumer 消费消息时，将 Header 的 organ-id 组织编号，添加到组织上下文。
+ * 7. Async：异步需要保证 ThreadLocal 的传递性，通过使用阿里开源的 TransmittableThreadLocal 实现。相关的改造点，可见：
+ *      1）Spring Async：
+ *          {@link com.cf.imes.framework.quartz.config.ChenfengAsyncAutoConfiguration#threadPoolTaskExecutorBeanPostProcessor()}
+ *      2）Spring Security：
+ *          TransmittableThreadLocalSecurityContextHolderStrategy
+ *          和 chenfengSecurityAutoConfiguration#securityContextHolderMethodInvokingFactoryBean() 方法
+ *
+ */
+package com.cf.imes.framework.organ;
@@ -0,0 +1,269 @@
+/*
+ * Copyright 2002-2021 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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 org.springframework.messaging.handler.invocation;
+
+import com.cf.imes.framework.organ.core.context.OrganContextHolder;
+import com.cf.imes.framework.organ.core.util.OrganUtils;
+import org.springframework.core.DefaultParameterNameDiscoverer;
+import org.springframework.core.MethodParameter;
+import org.springframework.core.ParameterNameDiscoverer;
+import org.springframework.core.ResolvableType;
+import org.springframework.lang.Nullable;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.HandlerMethod;
+import org.springframework.util.ObjectUtils;
+
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.lang.reflect.Type;
+import java.util.Arrays;
+
+import static com.cf.imes.framework.web.core.util.WebFrameworkUtils.HEADER_ORGAN_ID;
+
+/**
+ * Extension of {@link HandlerMethod} that invokes the underlying method with
+ * argument values resolved from the current HTTP request through a list of
+ * {@link HandlerMethodArgumentResolver}.
+ *
+ * 针对 rabbitmq-spring 和 kafka-spring，不存在合适的拓展点，可以实现 Consumer 消费前，读取 Header 中的 organ-id 设置到 {@link OrganContextHolder} 中
+ * TODO 晨丰：持续跟进，看看有没新的拓展点
+ *
+ * @author Rossen Stoyanchev
+ * @author Juergen Hoeller
+ * @since 4.0
+ */
+public class InvocableHandlerMethod extends HandlerMethod {
+
+    private static final Object[] EMPTY_ARGS = new Object[0];
+
+    private HandlerMethodArgumentResolverComposite resolvers = new HandlerMethodArgumentResolverComposite();
+
+    private ParameterNameDiscoverer parameterNameDiscoverer = new DefaultParameterNameDiscoverer();
+
+    /**
+     * Create an instance from a {@code HandlerMethod}.
+     */
+    public InvocableHandlerMethod(HandlerMethod handlerMethod) {
+        super(handlerMethod);
+    }
+
+    /**
+     * Create an instance from a bean instance and a method.
+     */
+    public InvocableHandlerMethod(Object bean, Method method) {
+        super(bean, method);
+    }
+
+    /**
+     * Construct a new handler method with the given bean instance, method name and parameters.
+     * @param bean the object bean
+     * @param methodName the method name
+     * @param parameterTypes the method parameter types
+     * @throws NoSuchMethodException when the method cannot be found
+     */
+    public InvocableHandlerMethod(Object bean, String methodName, Class<?>... parameterTypes)
+            throws NoSuchMethodException {
+
+        super(bean, methodName, parameterTypes);
+    }
+
+    /**
+     * Set {@link HandlerMethodArgumentResolver HandlerMethodArgumentResolvers} to use for resolving method argument values.
+     */
+    public void setMessageMethodArgumentResolvers(HandlerMethodArgumentResolverComposite argumentResolvers) {
+        this.resolvers = argumentResolvers;
+    }
+
+    /**
+     * Set the ParameterNameDiscoverer for resolving parameter names when needed
+     * (e.g. default request attribute name).
+     * <p>Default is a {@link DefaultParameterNameDiscoverer}.
+     */
+    public void setParameterNameDiscoverer(ParameterNameDiscoverer parameterNameDiscoverer) {
+        this.parameterNameDiscoverer = parameterNameDiscoverer;
+    }
+
+    /**
+     * Invoke the method after resolving its argument values in the context of the given message.
+     * <p>Argument values are commonly resolved through
+     * {@link HandlerMethodArgumentResolver HandlerMethodArgumentResolvers}.
+     * The {@code providedArgs} parameter however may supply argument values to be used directly,
+     * i.e. without argument resolution.
+     * <p>Delegates to {@link #getMethodArgumentValues} and calls {@link #doInvoke} with the
+     * resolved arguments.
+     * @param message the current message being processed
+     * @param providedArgs "given" arguments matched by type, not resolved
+     * @return the raw value returned by the invoked method
+     * @throws Exception raised if no suitable argument resolver can be found,
+     * or if the method raised an exception
+     * @see #getMethodArgumentValues
+     * @see #doInvoke
+     */
+    @Nullable
+    public Object invoke(Message<?> message, Object... providedArgs) throws Exception {
+        Object[] args = getMethodArgumentValues(message, providedArgs);
+        if (logger.isTraceEnabled()) {
+            logger.trace("Arguments: " + Arrays.toString(args));
+        }
+        // 注意：如下是本类的改动点！！！
+        // 情况一：无组织编号的情况
+        Long organId= parseorganId(message);
+        if (organId == null) {
+            return doInvoke(args);
+        }
+        // 情况二：有组织的情况下
+        return OrganUtils.execute(organId, () -> doInvoke(args));
+    }
+
+    private Long parseorganId(Message<?> message) {
+        Object organId = message.getHeaders().get(HEADER_ORGAN_ID);
+        if (organId == null) {
+            return null;
+        }
+        if (organId instanceof Long) {
+            return (Long) organId;
+        }
+        if (organId instanceof Number) {
+            return ((Number) organId).longValue();
+        }
+        if (organId instanceof String) {
+            return Long.parseLong((String) organId);
+        }
+        if (organId instanceof byte[]) {
+            return Long.parseLong(new String((byte[]) organId));
+        }
+        throw new IllegalArgumentException("未知的数据类型：" + organId);
+    }
+
+    /**
+     * Get the method argument values for the current message, checking the provided
+     * argument values and falling back to the configured argument resolvers.
+     * <p>The resulting array will be passed into {@link #doInvoke}.
+     * @since 5.1.2
+     */
+    protected Object[] getMethodArgumentValues(Message<?> message, Object... providedArgs) throws Exception {
+        MethodParameter[] parameters = getMethodParameters();
+        if (ObjectUtils.isEmpty(parameters)) {
+            return EMPTY_ARGS;
+        }
+
+        Object[] args = new Object[parameters.length];
+        for (int i = 0; i < parameters.length; i++) {
+            MethodParameter parameter = parameters[i];
+            parameter.initParameterNameDiscovery(this.parameterNameDiscoverer);
+            args[i] = findProvidedArgument(parameter, providedArgs);
+            if (args[i] != null) {
+                continue;
+            }
+            if (!this.resolvers.supportsParameter(parameter)) {
+                throw new MethodArgumentResolutionException(
+                        message, parameter, formatArgumentError(parameter, "No suitable resolver"));
+            }
+            try {
+                args[i] = this.resolvers.resolveArgument(parameter, message);
+            }
+            catch (Exception ex) {
+                // Leave stack trace for later, exception may actually be resolved and handled...
+                if (logger.isDebugEnabled()) {
+                    String exMsg = ex.getMessage();
+                    if (exMsg != null && !exMsg.contains(parameter.getExecutable().toGenericString())) {
+                        logger.debug(formatArgumentError(parameter, exMsg));
+                    }
+                }
+                throw ex;
+            }
+        }
+        return args;
+    }
+
+    /**
+     * Invoke the handler method with the given argument values.
+     */
+    @Nullable
+    protected Object doInvoke(Object... args) throws Exception {
+        try {
+            return getBridgedMethod().invoke(getBean(), args);
+        }
+        catch (IllegalArgumentException ex) {
+            assertTargetBean(getBridgedMethod(), getBean(), args);
+            String text = (ex.getMessage() != null ? ex.getMessage() : "Illegal argument");
+            throw new IllegalStateException(formatInvokeError(text, args), ex);
+        }
+        catch (InvocationTargetException ex) {
+            // Unwrap for HandlerExceptionResolvers ...
+            Throwable targetException = ex.getTargetException();
+            if (targetException instanceof RuntimeException) {
+                throw (RuntimeException) targetException;
+            }
+            else if (targetException instanceof Error) {
+                throw (Error) targetException;
+            }
+            else if (targetException instanceof Exception) {
+                throw (Exception) targetException;
+            }
+            else {
+                throw new IllegalStateException(formatInvokeError("Invocation failure", args), targetException);
+            }
+        }
+    }
+
+    MethodParameter getAsyncReturnValueType(@Nullable Object returnValue) {
+        return new AsyncResultMethodParameter(returnValue);
+    }
+
+    private class AsyncResultMethodParameter extends HandlerMethodParameter {
+
+        @Nullable
+        private final Object returnValue;
+
+        private final ResolvableType returnType;
+
+        public AsyncResultMethodParameter(@Nullable Object returnValue) {
+            super(-1);
+            this.returnValue = returnValue;
+            this.returnType = ResolvableType.forType(super.getGenericParameterType()).getGeneric();
+        }
+
+        protected AsyncResultMethodParameter(AsyncResultMethodParameter original) {
+            super(original);
+            this.returnValue = original.returnValue;
+            this.returnType = original.returnType;
+        }
+
+        @Override
+        public Class<?> getParameterType() {
+            if (this.returnValue != null) {
+                return this.returnValue.getClass();
+            }
+            if (!ResolvableType.NONE.equals(this.returnType)) {
+                return this.returnType.toClass();
+            }
+            return super.getParameterType();
+        }
+
+        @Override
+        public Type getGenericParameterType() {
+            return this.returnType.getType();
+        }
+
+        @Override
+        public AsyncResultMethodParameter clone() {
+            return new AsyncResultMethodParameter(this);
+        }
+    }
+
+}