租户解析器的camelContext属性鉴别器，使用jpa多租户和camel routeId

Question

我问你如何使用 camelContext 来获取事件触发的路由名称，更详细地说，我如何在 camelContext 中使用任何类型的鉴别器属性 x 进行谓词决策（如果 x =1 则 .. else . .) 例如：

我有这种路线：

//this route use the forst database
    from("direct:csvprocessor1")
        .routeId("tenant1")
        .from("file:src/main/resources/data/1?move=OUT&moveFailed=REFUSED")
        .unmarshal(csv)
        .to("bean:myCsvHandler?method=doHandleCsvData")
        .setBody(constant("OK VB"))
        .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(200))
        .setHeader(Exchange.CONTENT_TYPE, constant("text/html")); 

还有这条其他路线：

//this route use tenant2, the second database
    from("direct:csvprocessor1")
        .routeId("tenant2")
        .from("file:src/main/resources/data/2?move=OUT&moveFailed=REFUSED")
        .unmarshal(csv)
        .to("bean:myCsvHandler?method=doHandleCsvData")
        .setBody(constant("OK 2"))
        .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(200))
        .setHeader(Exchange.CONTENT_TYPE, constant("text/html"));

当我在 1 文件夹中拾取文件时，名为“tenant1”的第一条路线开始，在 2 中拾取文件时发生同样的情况，第二条路线tenant2 开始。它读取 csv 内容，内容必须使用 jpa 写入正确的租户X（数据库）

我必须在另一个类中检索 routeid 名称，但是这个类在骆驼上下文启动之前实例化，所以我不能注入上下文（因为这个类“BatchCurrentTenantIdentifierResolverImpl”属于 Spring 数据库初始化器）。我尝试添加方法“of”来设置camelContext，但我只得到tenant1，也是在路由2启动时，所以不能从租户切换到另一个租户（租户是数据库，我有两个数据库）：

@Component
public class BatchTenantContext {


    private static final Logger log = LoggerFactory.getLogger(BatchTenantContext.class);

    // don't Inject, use method Of because injecton was null
    CamelContext cctx;

    public BatchTenantContext(){getInstance();}

    private final static BatchTenantContext instance = new BatchTenantContext();

    public static BatchTenantContext getInstance(){
        return instance;
    }

    public synchronized String get() {
        if (cctx != null){
            Route val = cctx.getRoute("tenant1");
            if (val == null){
                val = cctx.getRoute("tenant2");
                if (val == null){
                    return "";
                }
                else {
                    return "tenant_2";
                }
            }
            else return "tenant_1";
        }
        return "";
    }

    public synchronized void of(CamelContext ctx){
        cctx = ctx;
    }

    public CamelContext getCamelContext()
    {
        return cctx;
    }


}

//multitenant approach, switch from one database to another 
//based on BatchTenantContext resolution..
    public class BatchCurrentTenantIdentifierResolverImpl implements CurrentTenantIdentifierResolver {

        static final Logger log = LoggerFactory.getLogger(BatchCurrentTenantIdentifierResolverImpl.class);


        @Override
        public String resolveCurrentTenantIdentifier() {
            String val = BatchTenantContext.getInstance().get();
            log.info("*** get tenant " + val);
            return val;
        }

        @Override
        public boolean validateExistingCurrentSessions() {
            return true;
        }

    }

那么，怎么知道路由火呢？请注意，上面的课程是单例的..我的方式正确吗？ 我使用 jpa whitin hibernate 提供程序，使用 rhe 多租户配置进行配置，如下所示：http://tech.asimio.net/2017/01/17/Multitenant-applications-using-Spring-Boot-JPA-Hibernate-and-Postgres.html 该应用程序在 spring-boot 运行时环境或 Tomcat 应用服务器中工作。

关于所有的任何想法？

非常感谢！

罗比

我添加了这段代码：

    @Configuration
    @EnableConfigurationProperties({ MultiTenantAfSissProperties.class, JpaProperties.class })
    @ImportResource(locations = { "classpath:applicationContent.xml" })
    @EnableTransactionManagement
    public class MultiTenantJpaConfiguration {

        static final Logger logger = LoggerFactory.getLogger(MultiTenantJpaConfiguration.class);

        @Inject
        private JpaProperties jpaProperties;

        @Inject
        MultiTenantAFSISSProperties multiTenantAFSISSProperties; //lista dei datasources collegati ai tenant

        @Bean
        public Map<String, DataSource> dataSourceRetrieval(){
            Map<String, DataSource> result = new HashMap<>();
            for (DataSourceProperties dsProperties : this.multiTenantAFSISSProperties.getDataSources()) {
                DataSourceBuilder factory = DataSourceBuilder
                    .create()
                    .url(dsProperties.getUrl())
                    .username(dsProperties.getUsername())
                    .password(dsProperties.getPassword())
                    .driverClassName(dsProperties.getDriverClassName());
                result.put(dsProperties.getTenantId(), factory.build());
            }
            return result;
        }

        /**
         * 
         * @return
         */
        @Bean
        public MultiTenantConnectionProvider multiTenantConnectionProvider(){
            return new AfsissMultiTenantConnectionProviderImpl();
        }

        /**
         * 
         * @return
         */
        @Bean 
        public CurrentTenantIdentifierResolver currentTenantIdentifierResolver(){
            return new BatchCurrentTenantIdentifierResolverImpl();
        }

        /**
         *  
         * @param multiTenantConnectionProvider
         * @param currentTenantIdentifierResolver
         * @return
         */
        @Bean
        public LocalContainerEntityManagerFactoryBean entityManagerFactoryBean(MultiTenantConnectionProvider multiTenantConnectionProvider,
            CurrentTenantIdentifierResolver currentTenantIdentifierResolver) {

            Map<String, Object> hibernateProps = new LinkedHashMap<>();
            hibernateProps.putAll(this.jpaProperties.getProperties());

            Map<String,String> all = this.jpaProperties.getProperties();
            for ( Map.Entry<String, String> prop : all.entrySet()){
                System.out.println(" " + prop.getKey() + " = " + prop.getValue());
            }

            hibernateProps.put(Environment.MULTI_TENANT, MultiTenancyStrategy.DATABASE);
            hibernateProps.put(Environment.MULTI_TENANT_CONNECTION_PROVIDER, multiTenantConnectionProvider);
            hibernateProps.put(Environment.MULTI_TENANT_IDENTIFIER_RESOLVER, currentTenantIdentifierResolver);

            // No dataSource is set to resulting entityManagerFactoryBean
            LocalContainerEntityManagerFactoryBean result = new LocalContainerEntityManagerFactoryBean();
            result.setPackagesToScan(new String[] { AfFileEntity.class.getPackage().getName() }); 
            result.setJpaVendorAdapter(new HibernateJpaVendorAdapter());
            result.setJpaPropertyMap(hibernateProps);

            return result;
        }

        /**
         * crea la factory per ricavare l'entity manager
         * @param entityManagerFactoryBean
         * @return
         */
        @Bean
        public EntityManagerFactory entityManagerFactory(LocalContainerEntityManagerFactoryBean entityManagerFactoryBean) {
            return entityManagerFactoryBean.getObject();
        }

        /**
         * get transaction manager 
* @param entityManagerFactory
         * @return
         */
        @Bean
        public PlatformTransactionManager txManager(EntityManagerFactory entityManagerFactory) {
            SessionFactory sessionFactory = entityManagerFactory.unwrap(SessionFactory.class);
            HibernateTransactionManager result = new HibernateTransactionManager();
            result.setAutodetectDataSource(false);
            result.setSessionFactory(sessionFactory);
            return result;
        }



}

在 applicationContent.xml 中：

<jpa:repositories base-package="com.xxx.dao" transaction-manager-ref="txManager" />
<tx:annotation-driven transaction-manager="txManager" proxy-target-class="true" />

每次我在 csvHanlder 中使用实体管理器和事务管理器时，spring 事务管理器都会在上面的 currentTenantIdentifierResolver() 方法中调用 BatchCurrentTenantIdentifierResolverImpl 类：

@Component
@Transactional(propagation = Propagation.REQUIRED)
public class MyCsvHandler {

    @Inject
    AFMOVCrudRepository _entitymanagerMov; //it extends JpaRepository

    @Inject
    AFVINCCrudRepository _entityManagerVINC;//it extends JpaRepository


    @Inject
    AFFileCrudRepository _entityManagerAfFile;//it extends JpaRepository




    static final Logger logger = LoggerFactory.getLogger(MyCsvHandler.class);



//save csv data on the right table on the right tenant
    public void doHandleCsvData(List<List<String>> csvData) throws FileNotEvaluableException
    {
        //System.out.println("stampo..");
        boolean status = true;
            if (csvData.size() > 0){
            AfFileEntity afbean = new AfFileEntity();
            afbean.setNomeFile("test");
            afbean.setDataImport(new java.sql.Timestamp(System.currentTimeMillis()));
            afbean.setTipoFile("M");
            afbean.setAfStatoFlusso("I");

            _entityManagerAfFile.save(afbean);

            long pkfile = afbean.getId();
            logger.info("pkfile: " + pkfile);

            int i = 1; 
            logger.info("file size:" + csvData.size());
            for (List<String> rows : csvData){

                //for (int j = 0; i < rows.size(); j++){
                if (rows.get(2).trim().equalsIgnoreCase(...)){ 
                    MovEntity mbean = new MovEntity();
                    setMovFields(mbean, rows);
                    mbean.setAfFileId(afbean);
                    logger.info(String.valueOf((i++)) +  " " + mbean);

                    _entitymanagerMov.save(mbean);
                }
                else if (rows.get(2).trim().equalsIgnoreCase(..) || rows.get(2).trim().equalsIgnoreCase(..) ) { 
                    VincEntity vincBean = new VincEntity();  
                    setVincFields(vincBean, rows);
                    vincBean.setAfFileId(afbean);

                    logger.info(String.valueOf((i++)) +  " " + vincBean);

                    _entityManagerVINC.save(vincBean);
                }   
                else {
                    status = false;
                    break;
                }
            }
             if (!status) throw new FileNotEvaluableException("error file format");
        }

    }

    private void setVincFields(VincEntity vincBean, List<String> rows) {
        vincBean.setXXX().. and others methods
    }

    private void setMovFields(MovEntity mbean, List<String> rows) {
        mbean.setStxxx() and other .. methods

    }
            return new 

Answer 1

您的路线中有这样的东西

from("direct:csvprocessor1").routeId("tenant2").process((Exchange e)-> {
                BatchCurrentTenantIdentifierResolverImpl.tenant.set("tenant_1");
            })
        .from("file:src/main/resources/data/2?move=OUT&moveFailed=REFUSED")
        .unmarshal().csv()
        .to("bean:myCsvHandler?method=doHandleCsvData")
        .setBody(constant("OK 2"))
        .setHeader(Exchange.HTTP_RESPONSE_CODE, constant(200))
        .setHeader(Exchange.CONTENT_TYPE, constant("text/html"));

并在您的BatchCurrentTenantIdentifierResolverImpl 中将其实现为
public

class BatchCurrentTenantIdentifierResolverImpl {
    public static ThreadLocal<String> tenant = new ThreadLocal<String>();
    static final Logger log = LoggerFactory.getLogger(BatchCurrentTenantIdentifierResolverImpl.class);


    @Override
    public String resolveCurrentTenantIdentifier() {
        String val = tenant.get();
        log.info("*** get tenant " + val);
        return val;
    }

    @Override
    public boolean validateExistingCurrentSessions() {
        return true;
    }       
}