我听说过很多关于 Apache Camel 与 Fuse 的成功集成故事。因此。我刚刚开始探索 Fuse IDE，脑子里只有一个简单的任务，我想实现:

1. 读取固定长度的文件
2. 解析定长文件
3. 持久化到mysql数据库表

我只能做到:

1. 读取定长文件(端点为“file:src/data/Japan?noop=true”)
2. 使用 Bindy 定义一个 Marshal 并使用 @FixedLengthRecord 注解定义一个 POJO 包模型
3. 然后我卡住了...如何将 POJO 持久化到 mysql 数据库表中？我可以看到一些 JDBC、IBatis 和 JPA 端点，但如何在 Fuse IDE 中实现这些端点？

My POJO package:
package com.mbww.model;

import org.apache.camel.dataformat.bindy.annotation.DataField;
import org.apache.camel.dataformat.bindy.annotation.FixedLengthRecord;

@FixedLengthRecord(length=91)
public class Japan {

@DataField(pos=1, length=10)
private String TNR; 

@DataField(pos=11, length=10)
private String ATR;

@DataField(pos=21, length=70)
private String STR; 
}

最佳答案

您可以使用以下所有组件来实际读取和写入数据库:

1. JDBC
2. IBATIS
3. MyBATIS
4. SPRING-JDBC
5. SQL
6. 定制处理器

我将向您展示如何使用自定义处理器将行插入表中。这样做的主要原因是您将开始处理消息和交换，这将使您对 Camel 有更多的了解。可以按照 camel 站点上的文档使用所有其他组件。

那么让我们回顾一下你有什么。您正在读取文件并将正文转换为绑定(bind)对象。因此，对于文本文件中的每一行，Camel 都会发送一个 com.mbww.model.JAPAN 类的绑定(bind)对象。到下一个终点。下一个端点需要与数据库对话。有一个问题我可以立即发现你正在使用 marshal你应该使用 unmarshal .

文档明确指出:如果您从 Camel 组件之一(例如文件、HTTP 或 JMS)接收到一条消息，您通常希望将有效负载解码到某个 bean 中，以便您可以使用一些 Bean Integration 或执行谓词评估等。为此，请使用 Java 中 DSL 或 Xml 配置中的 unmarshal 词。。

您的 bindy 类看起来不错，但它缺少 getter 和 setter 将类修改为如下所示:

package com.mbww.model;
import org.apache.camel.dataformat.bindy.annotation.DataField;
import org.apache.camel.dataformat.bindy.annotation.FixedLengthRecord;

@FixedLengthRecord(length=91)
public class Japan {

  @DataField(pos=1, length=10)
  private String TNR; 

  @DataField(pos=11, length=10)
  private String ATR;

  @DataField(pos=21, length=70)
  private String STR;

  public String getTNR() {
    return TNR;
  }

  public void setTNR(String tNR) {
    TNR = tNR;
  }

  public String getATR() {
    return ATR;
  }

  public void setATR(String aTR) {
    ATR = aTR;
  }

  public String getSTR() {
    return STR;
  }

  public void setSTR(String sTR) {
    STR = sTR;
  } 
}

首先，您需要在路由中为数据库创建一个数据源。首先是将 mysql 驱动程序 jar 添加到您的 Maven 依赖项中打开您的 pom.xml文件并向其添加以下依赖项。

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <!-- use this version of the driver or a later version of the driver -->
    <version>5.1.25</version>
</dependency>

现在我们需要声明一个自定义处理器以在将使用此驱动程序的路由中使用，并将接收到的正文插入到表中。

所以让我们在 Fuse IDE 中创建一个名为 PersistToDatabase 的新类代码如下:

package com.mbww.JapanData;
import java.sql.DriverManager;
import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Map;

import org.apache.camel.Body;
import org.apache.camel.Exchange;
import org.apache.camel.Handler;
import org.apache.camel.Headers;

import com.mbww.model.Japan;
import com.mysql.jdbc.Statement;


public class PersistToDatabase {
@Handler
public void PersistRecord
(
        @Body Japan msgBody
        , @Headers Map hdr
        , Exchange exch
) throws Exception
{
    try {
        Class.forName("com.mysql.jdbc.Driver");
    } catch (ClassNotFoundException e) {
        System.out.println("Where is your MySQL JDBC Driver?");
        e.printStackTrace();
        return;
    }

    System.out.println("MySQL JDBC Driver Registered!");
    Connection connection = null;

    try {
        connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/databasename","root", "password");

    } catch (SQLException e) {
        System.out.println("Connection Failed! Check output console");
        e.printStackTrace();
        return;
    }

    if (connection != null) {
        System.out.println("You made it, take control your database now!");
    } else {
        System.out.println("Failed to make connection!");
    }
    try {
        PreparedStatement stmt=connection.prepareStatement("INSERT INTO JapanDate(TNR,ATR,STR) VALUES(?,?,?)");
        stmt.setString(1, msgBody.getTNR());
        stmt.setString(2, msgBody.getATR());
        stmt.setString(1, msgBody.getSTR());
        int rows = stmt.executeUpdate();
        System.out.println("Number of rows inserted: "+Integer.toString(rows));
    }
    catch(Exception e){
        System.out.println("Error in executing sql statement: "+e.getMessage() );
        throw new Exception(e.getMessage());
    }

}
}

这个类是一个 POJO，除了 @Handler 之外没什么特别的PersistRecord 上的注释.这个注解告诉 Camel PersistRecord方法/过程将处理消息交换。您还会注意到方法 PersistRecord有一个日本类型的参数。如前所述，当您在 Camel route 调用转换 bean 时，它会将每一行转换为日本对象并沿路线传递。

剩下的代码只是如何处理 JDBC 连接和调用插入语句。

我们几乎完成了最后一件事。我们需要在我们的 Camel 路线 xml 中声明这个类。根据您的架构类型，此文件通常称为 camel-route.xml 或 blueprint.xml。打开源选项卡并添加以下行 <bean id="JapanPersist" class="com.mbww.JapanData.PersistToDatabase"/>在 <camelContext> 之前标签。

这声明了一个名为 JapanPersist 的新 spring bean基于我们刚刚添加到 Camel 路线的类(class)。您现在可以在 Camel route 引用此 bean。

因此最终的路由 xml 文件应该如下所示:

<?xml version="1.0" encoding="UTF-8"?>
<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"
   xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
   xmlns:camel="http://camel.apache.org/schema/blueprint"
   xsi:schemaLocation="
   http://www.osgi.org/xmlns/blueprint/v1.0.0 http://www.osgi.org/xmlns/blueprint/v1.0.0/blueprint.xsd
   http://camel.apache.org/schema/blueprint http://camel.apache.org/schema/blueprint/camel-blueprint.xsd">

<bean id="JapanPersist" class="com.mbww.JapanData.PersistToDatabase"/>

  <camelContext trace="false" id="blueprintContext" xmlns="http://camel.apache.org/schema/blueprint">
    <route id="JapanDataFromFileToDB">
        <from uri="file:src/data/japan"/>
        <unmarshal ref="Japan"/>
        <bean ref="JapanPersist"/>        
    </route>
</camelContext>

</blueprint>

或者看下面的截图:

一旦理解了这项技术，您就可以开始使用拆分器、连接池和线程来扩展解决方案，以执行大量并发插入等操作。

使用上面的技术，您了解了如何将自己的 bean 注入(inject)到 Camel 路由中，这使您能够直接在代码中处理消息。

我没有测试过代码，所以可能会有一两个错误，但思路应该很清楚。

关于mysql - Fuse ide如何定义数据库表端点，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/23706429/