使用Apache Camel同步FTP remote和FS（或HDFS）

Question

有时，我想手动运行一个工具，将FTP远程与我的FS（至少，最终-HDFS）远程同步。每个文件都应该下载，更新的文件应该追加。

我使用Apache Camel FTP2示例:https://github.com/Apache/Camel/tree/master/examples/camel-example-FTP启动该项目。在其他来源的帮助下，这是一个简单的解决方案。

object Main{

  def main(args: Array[String]): Unit = {
    val context = new DefaultCamelContext
    context.addRoutes(FtpRoute())

    context.start
    Thread.sleep(100000)
    context.stop
  }    
}

case class FtpRoute() extends RouteBuilder {

  def configure(): Unit = { // configure properties component
    val pc = getContext.getComponent("properties", classOf[PropertiesComponent])
    pc.setLocation("classpath:ftp.properties")

    val ftpSource = getContext.resolvePropertyPlaceholders(
      s"ftp://{{ftp.serv}}{{ftp.path}}?username={{ftp.user}}&password={{ftp.pass}}&passiveMode=true")

    from(ftpSource)
      .to("file:/tmp/ftp")
      .log("Downloaded file ${file:name} complete.")
  }
}

我可以说这很管用，但是...绝对不是我想要的。

null

多谢！

Answer 1

ftp组件正在轮询使用者。

这样的组件每延迟一次就轮询一个URI。

每个轮询，组件从第一个到最后接收所有新文件。这就是为什么你会看到文件不断的重新下载。下载完所有文件后，组件将停止，并将等待计时器或计划发出的新事件。

null

disconnect=true&
connectTimeout=300000&
ftpClient.dataTimeout=300000&
timeout=300000&soTimeout=300000&
delay=3600000&reconnectDelay=10000&
backoffErrorThreshold=1&
maximumReconnectAttempts=3&
backoffMultiplier=2&
passiveMode=true&
noop=true&
idempotent=true&
idempotentRepository=#jdbcDocFileIdempotentRepository&
download=true&
recursive=true&
stepwise=true&
antInclude=**/currMonth/*.zip&antExclude=**/prevMonth/*.zip&
idempotentKey=${file:onlyname}-${file:size}

<bean id="jdbcDocFileIdempotentRepository"
      class="org.apache.camel.processor.idempotent.jdbc.JdbcMessageIdRepository">
    <argument ref="embeddedDataSource"/>
    <argument value="DocFileRepo"/>
    <property name="tableExistsString" value="SELECT 1 FROM MESSAGE_REPOSITORY WHERE 1 = 0"/>
    <property name="createString" value="CREATE TABLE MESSAGE_REPOSITORY (processorName TEXT, messageId TEXT, createdAt TIMESTAMP, result TEXT)"/>
    <property name="queryString" value="SELECT COUNT(*) FROM MESSAGE_REPOSITORY WHERE processorName = ? AND messageId = ?"/>
    <property name="insertString" value="INSERT INTO MESSAGE_REPOSITORY (processorName, messageId, createdAt) VALUES (?, ?, ?)"/>
    <property name="deleteString" value="DELETE FROM MESSAGE_REPOSITORY WHERE processorName = ? AND messageId = ?"/>
</bean>

数据源：

<bean id="embeddedDataSource" class="org.sqlite.javax.SQLiteConnectionPoolDataSource">
   <property name="url" value="jdbc:sqlite:${embedded_db_config_path}" />
   <property name="encoding" value="UTF-8"/>
   <property name="journalMode" value="WAL"/>
</bean>

已更新

根据Camel文档：