java - Camel jdbc : How can I reset datasource if mysql connection gets closed

Question

我们开发了一个 Camel bundle(部署在 Karaf 中)，预计每 24 小时从 MySQL 中拉取数据并推送到 S3。但是，如果 MySQL 闲置 8 小时，它会在内部关闭连接，因此在下一次计划执行时，它会开始抛出错误。请查看我们代码中的以下片段。

属性:

MySqlDriver=com.mysql.jdbc.Driver
MySqlDatabaseURL=jdbc:mysql://x.x.x.x/dbname?autoReconnect=true
MySqlUsername=sm*****
MySqlPassword=*******

激活剂:

public class Activator implements BundleActivator {

    public CamelContext context = null;

    public void start(BundleContext bundleContext) throws Exception {
        DataSource dataSource = UDMSUtils.createDataSource(UDMSUtils.getProperty(UDMSConstants.MYSQL_DATABASE_URL));

        SimpleRegistry simpleRegistry = new SimpleRegistry();
        simpleRegistry.put(UDMSConstants.UDMS_DATA_SOURCE, dataSource);

        context = new OsgiDefaultCamelContext(bundleContext, simpleRegistry);
        context.addRoutes(new CreativeRoutes());
        context.start();
    }

}

构建数据源:

public static DataSource createDataSource(String connectURI) {
    BasicDataSource ds = new BasicDataSource();
    ds.setDriverClassName(getProperty(UDMSConstants.MYSQL_DRIVER));
    ds.setUsername(getProperty(UDMSConstants.MYSQL_USERNAME));
    ds.setPassword(getProperty(UDMSConstants.MYSQL_PASSWORD));
    ds.setUrl(connectURI);
    ds.setMaxWait(-1);  // Waits indefinately
    return ds;
}

路线:

from("timer://Timer?repeatCount=1").to("direct:record_count").end();

from("direct:record_count")
    .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
            exchange.getIn().setBody(query);
        }
    })    
    .routeId("record_count")
    .to("jdbc:" + UDMSConstants.UDMS_DATA_SOURCE)
    .process(new Processor() {
        @Override
        public void process(Exchange exchange) throws Exception {
            // ...
        }
    );

任何人都可以建议，需要对上述代码进行哪些更改，以便连接在我们需要的时间内保持 Activity 状态。

请注意:我们无权更改 mysql.properties，因此我们需要在代码中处理此问题。

Answer 1

我之前遇到过类似的问题。 VikingSteve 也很准确地建议您去做。因为我使用的是 OSGI 蓝图，所以我用 XML 完成了所有配置，所以我着手解决它如下。

1) 将 Apache Commons DBCP 依赖项添加到您的 pom:

<dependency>
    <groupId>commons-dbcp</groupId>
    <artifactId>commons-dbcp</artifactId>
    <version>1.4</version>
</dependency>

2) 在你的 camel 路由/蓝图文件中声明连接池如下:

<bean id="MydataSource" class="org.apache.commons.dbcp.BasicDataSource" destroy-method="close" scope="singleton" >
    <property name="driverClassName" value="com.mysql.jdbc.Driver"/>
    <property name="url" value="jdbc:mysql://DB-001:3306/Customer"/>
    <property name="username" value="sys_ETL"/>
    <property name="password" value="Blah"/>
    <property name="initialSize" value="4"/>
    <property name="maxActive" value="32"/>
    <property name="maxIdle" value="16"/>
    <property name="minIdle" value="8"/>
    <property name="timeBetweenEvictionRunsMillis" value="1800"/>
    <property name="minEvictableIdleTimeMillis" value="1800"/>
    <property name="testOnBorrow" value="true"/>
    <property name="testWhileIdle" value="true"/>
    <property name="testOnReturn" value="true"/>
    <property name="validationQuery" value="SELECT 1"/>
    <property name="maxWait"  value="1000"/>
    <property name="removeAbandoned" value="true"/>
    <property name="logAbandoned" value="true"/>
    <property name="removeAbandonedTimeout" value="30000"/>
</bean>

此步骤将创建一个数据库连接池作为一个 bean，然后我可以在我的路由中使用它。这个 bean 的名称是 Mydatasource 我稍后会用到这个信息。还要注意我在配置中为连接池设置的属性。这些属性允许我的连接池增长和收缩，而且它确保即使在空闲之后连接也不会失效。

3) 创建一个POJO来使用这个连接池:

public class AccountInformationToDatabase {


private BasicDataSource dataSource;
public BasicDataSource getDataSource() {
    return dataSource;
}
public void setDataSource(BasicDataSource dataSource) {
    this.dataSource = dataSource;
}
@Handler
public void PersistRecord
(
        @Body AccountRecordBindy msgBody
        , @Headers Map hdr
        , Exchange exch
) throws Exception
{

    Connection conn = null;
    PreparedStatement stmt=null;



    try 
    {


        conn= dataSource.getConnection();
        stmt =conn.prepareStatement("SOME INSERT STATEMENT");   

        stmt.setString(1,msgBody.getAccountNumber().trim());
        stmt.setString(2,msgBody.getRecordType().trim() );
        stmt.setString(3,msgBody.getSequenceNumber().trim());
        stmt.setString(4,msgBody.getTitle().trim());
        stmt.setString(5,msgBody.getCustomerType().trim());
        stmt.setString(6,msgBody.getName().trim());
        stmt.setString(7,msgBody.getAccountAddress1().trim());


        stmt.executeUpdate();        






    }
    catch (Exception e)
    {

        throw new Exception(e.getMessage());

    }

    finally
    {
        try
        {
                if (stmt!=null)
                {
                    stmt.close();
                    stmt= null;
                }
                if (conn!=null)
                {
                    conn.close();
                    conn= null;
                }
        }
        catch(SQLException e)
        {

            throw new Exception(e.getMessage());

        }

    }


}

这个 POJO 有一个名为数据源的属性，类型为 org.apache.commons.dbcp.BasicDataSource。我现在可以将 Mydatasource bean 注入(inject)到这个 POJO 中，这样我的类就可以访问连接池了。

4) 将 POJO 变成一个 bean 并注入(inject)连接池:

<bean id="AccountPersist"   class="AccountInformationToDatabase">
    <property name="dataSource" ref="MydataSource"/>
</bean>

如果您正在处理文本文件并希望使用并发插入等，则此技术是必须具备的。