运行spring boot调度程序和Apache Camel时出现问题

Question

我正在尝试一个使用spring boot和Apache骆驼文件组件的演示文件传输程序。我用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();
    }

}

当我运行程序时，调度程序正在运行三次，一个接一个地传输三个文件。此后，排定程序不再运行。然后，当我试图通过关闭嵌入的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.我如何在第一次运行时同时传输所有的三个文件，而不是一个一个地传输？

Answer 1

关于问题3：我认为这是不可能的，请参阅其他相关问题中的评论

对于问题#1和#2:理解这些行为的关键是，您在接收模式下使用pollrich，因为您没有设置任何超时：请参阅Pollrich>文档中关于超时的良好解释，特别是：

使用超时值的良好做法===默认情况下，Camel将使用receive。直到有可用的消息为止。因此，建议始终提供一个超时值，以明确我们可以等待消息，直到达到超时值。===

要回答您的问题：以下是当前路由设置的情况：

>

spring调度程序每60秒通过向“direct：transferfile”通道发送一条正文为“null”的虚拟消息来触发FileTransferRoute

FileTransferRoute检测并使用此消息：已创建交换，将步骤#1中创建的消息作为输入消息，

FileTransferRoute中的

PollRich组件将阻塞，直到在输入目录中检测到新文件为止（因为未定义超时）

=>这还会阻止spring调度程序，因为您使用的是同步方法ProducerTemplate.SendBody()和Direction:通道(Direction组件与同步调用一起工作）

=>这回答了您的第一个问题：spring调度程序实际上仍在运行，但它的线程在sendbody()方法调用中被阻塞，这是因为pollrich组件轮询新传入文件。输入目录中的下一个传入文件将恢复轮询；排定的方法应该完成，并在60秒后触发一个新的调用。排定程序没有理由在第一个文件处理完毕后停止。

当您停止springboot应用程序时，pollrich组件仍然被阻止，轮询新的传入文件：这意味着仍有一个挂起的交换（在上面的步骤2中创建）尚未完成

=>这回答了您的第二个问题：“Inflight Exchange”实际上是由最新调度程序调用创建的最新Exchange。Camel默认关机策略将在关机前等待X秒，以防挂起（飞行中）交换。

我建议您为骆驼记录器启用调试级别，您将更好地理解引擎盖下发生的事情。

只需在pollrelch组件中使用超时，您就可以解决飞行交换问题并改进您的航线。由于您是在专用的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();

希望这有帮助。