租户解析器的camelContext属性鉴别器，使用jpa多租户和camel routeId
Posted
技术标签:
【中文标题】租户解析器的camelContext属性鉴别器，使用jpa多租户和camel routeId【英文标题】：camelContext attribute discriminator for tenant resolver , using jpa multitenant and camel routeId 【发布时间】：2018-02-06 13:22:10 【问题描述】：我问你如何使用 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
【问题讨论】：
您是否希望使用BatchTenantContext 中的同步获取方法获取“当前正在执行的路由”？那是行不通的。你总是得到tenant_1，因为这就是你实现get方法的方式。 CamelContext.getRoute("id") 退出时返回路由。在您的情况下，两条路线都存在。如果您反转代码并首先检查tenant_2，您将始终得到它。它不会返回当前正在执行的路由。
@Gautam ，是的，我希望获得当前正在执行的路线，但我还没有弄清楚如何......有人可以帮我一些建议吗？谢谢
如果你能更好地解释你的流程。我在这里的理解是，您有两个目录 1 和 2，并且根据文件到达那里的时间，触发您的一条路线。该路由调用myCsvHandler，您可能正在尝试将数据插入两个租户之一。 BatchCurrentTenantIdentifierResolverImpl 究竟是如何使用的？ bean myCsvHandler 是否调用此类上的方法来确定当前租户？在您的第一个路由中，将消息头添加为租户_1，在第二个路由中添加一个租户_2。然后在你的 bean myCsvHandler 中使用 Exchange 中的那个。
@Gautam 我为您的问题添加代码。对 BatchCurrentTenantIdentifierResolverImpl 的应用程序调用发生在 myCsvHanlder 方法调用之前。我发布的链接使用 threadlocal 来获取上下文，因为它是一个 Web servlet 应用程序。但是我的路由不是休息或休息调用...所以我必须保留其他方法的上下文缓存..但我不是骆驼专家..
这个BatchCurrentTenantIdentifierResolverImpl调用什么时候发生在路由中？在处理路由时还调用了这个类的什么方法？在您的路线中看不到。
【参考方案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;
【讨论】：
以上是关于租户解析器的camelContext属性鉴别器，使用jpa多租户和camel routeId的主要内容，如果未能解决你的问题，请参考以下文章
EF- '' 和 '' 都映射到表 ''。层次结构中没有鉴别器的所有实体类型必须映射到不同的表
Spirit X3：替代解析器的属性，不是`char`，而是`variant<char, char>`