首页 > 解决方案 > 运行 Spring Boot Scheduler 和 Apache Camel 时出现问题

问题描述

我正在尝试使用 Spring Boot 和 Apache Camel 文件组件的演示文件传输程序。我使用 Spring Boot 编写了一个调度程序,它将每 1 分钟运行一次,它调用 Apache Camel 路由并进行文件传输。我在目录 C:\CamelDemo\inputFolder 中有三个文件,即 input1.txt、input2.txt 和 input3.txt。我的 Spring Boot 调度程序如下:

package com.example.demo.scheduler;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.camel.ProducerTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class Scheduler {

    @Autowired private ProducerTemplate producerTemplate;

    @Scheduled(fixedRate=60000)
    public void fixedRateSch() {
          SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
          Date now = new Date();
          String strDate = sdf.format(now);
          System.out.println("Fixed Rate scheduler:: " + strDate);


          producerTemplate.sendBody("direct:transferFile", null);
          System.out.println("Success");

       }

}

我的路线如下:

package com.example.demo.route;

import org.apache.camel.LoggingLevel;
import org.apache.camel.builder.RouteBuilder;
import org.springframework.stereotype.Component;

@Component
public class FileTransferRoute extends RouteBuilder {

    @SuppressWarnings("deprecation")
    @Override
    public void configure() {
        errorHandler(defaultErrorHandler()
            .maximumRedeliveries(3)
            .redeliverDelay(1000)
            .retryAttemptedLogLevel(LoggingLevel.WARN));

        from("direct:transferFile")
            .log("Route reached")
            .pollEnrich("file:C:\\CamelDemo\\inputFolder?noop=true")
            .to("file:C:\\CamelDemo\\outputFolder?autoCreate=false")
        .end();
    }

}

当我运行程序时,调度程序正在运行 3 次,一个接一个地传输三个文件。之后调度程序不再运行。然后,当我试图通过关闭嵌入式 Tomcal 来停止 Spring Boot 应用程序时,它会给出以下错误:

2019-09-11 15:51:14.711  INFO 10408 --- [n(15)-127.0.0.1] inMXBeanRegistrar$SpringApplicationAdmin : Application shutdown requested.
2019-09-11 15:51:14.714  INFO 10408 --- [n(15)-127.0.0.1] o.a.camel.spring.SpringCamelContext      : Apache Camel 2.24.0 (CamelContext: camel-1) is shutting down
2019-09-11 15:51:14.714  INFO 10408 --- [n(15)-127.0.0.1] o.a.camel.impl.DefaultShutdownStrategy   : Starting to graceful shutdown 1 routes (timeout 300 seconds)
2019-09-11 15:51:14.717  INFO 10408 --- [ - ShutdownTask] o.a.camel.impl.DefaultShutdownStrategy   : Waiting as there are still 1 inflight and pending exchanges to complete, timeout in 300 seconds. Inflights per route: [route1 = 1]
2019-09-11 15:51:14.718  INFO 10408 --- [ - ShutdownTask] o.a.camel.impl.DefaultShutdownStrategy   : There are 1 inflight exchanges:
    InflightExchange: [exchangeId=ID-PCIN467166-1568196927146-0-10, fromRouteId=route1, routeId=route1, nodeId=pollEnrich1, elapsed=0, duration=167106]

所以有以下问题: 1. 我怎样才能让调度程序永远运行,以便它继续轮询文件位置,当文件夹中出现新文件时,它会将文件传输到输出目录。2. 如何正确关闭我想要的 Spring Boot 应用程序,为什么在关闭过程中会抛出错误?3. 如何在第一次运行时同时传输所有三个文件而不是一个一个地传输?

标签: spring-bootapache-camelscheduler

解决方案


对于问题 #3:我认为这是不可能的,请参阅您其他相关问题中的评论

对于问题 #1 和 #2:了解此行为的关键是您使用的pollEnrich接收模式,因为您没有设置任何超时:请参阅PollEnrich 文档中有关超时的良好解释,特别是:

使用超时值的好习惯 === 默认情况下,Camel 将使用接收。在有可用消息之前,这可能会阻塞。因此,建议始终提供超时值,以明确表示我们可能会等待消息,直到超时。===

回答您的问题:您当前的路线设置如下:

  1. 每 60 秒,Spring 调度程序FileTransferRoute通过向“direct:transferFile”通道发送一条带有“null”正文的虚拟消息来触发
  2. FileTransferRoute检测并使用此消息:已创建一个 Exchange,将在步骤 #1 中创建的消息作为输入消息,
  3. pollEnrich块中的组件,FileTransferRoute直到在输入目录中检测到新文件(因为没有定义超时)

    => 这也将阻塞 Spring 调度程序,因为您正在使用同步方法producerTemplate.sendBody()direct:通道(direct组件与同步调用一起使用)

    => 这回答了您的第一个问题:Spring 调度程序实际上仍在运行,但它的线程在sendBody()方法调用中被阻塞,因为pollEnrich组件轮询新的传入文件。输入目录中的下一个传入文件将恢复轮询;然后计划的方法应该完成,并且应该在 60 秒后触发新的调用。在处理完第一个文件后,调度程序没有理由停止。

  4. 当您停止 Springboot 应用程序时,该pollEnrich组件仍然被阻塞,轮询新的传入文件:这意味着仍有一个未完成的 Exchange(在上面的步骤 2 中创建)尚未完成

    =>这回答了你的第二个问题:“飞行中的交换”实际上是由最新的调度程序调用创建的最新的交换。Camel 默认关闭策略将在关闭前等待 X 秒,以防挂起(飞行中)交换。

我建议您为 Camel 记录器启用 DEBUG 级别,您将更好地了解幕后发生的事情。

您可以通过在 pollEnrich 组件中使用超时来摆脱飞行中的交换问题并改进您的路线。由于您在专用的 Spring 调度程序中实现调度程序逻辑,我认为您应该使用receiveNoWait模式(0s 超时)。

    from("direct:transferFile").autoStartup(true)
            .pollEnrich("file:C:\\var\\CamelDemo\\inputFolder?noop=true&delay=2000", 0)

            .choice()
                .when(exchange -> exchange.getIn().getBody() != null)
                    .to("file:C:\\var\\CamelDemo\\outputFolder?autoCreate=false")
                // handle case where no file is found 
                .otherwise()
                    .log(" skip, no file found. ")
            .end();

希望这可以帮助。


推荐阅读