Apache Camel 路由超时

Question

我有两条骆驼路线，用 XML 配置并粘贴在下面：-

路线 1：

<camel:route id="statementsArchivingPollRoute">
    <camel:from uri="timer://tempQueue?fixedRate=true&amp;period=30s&amp;delay=30s"/>
    <camel:transacted ref="PROPAGATION_REQUIRED">
    <camel:process ref="statementsArchivingRequestZipProcessor"/>
    <camel:choice>
        <camel:when>
            <camel:simple>${body.size} >= 1</camel:simple>
            <camel:split>
                <camel:simple>${body}</camel:simple>
                <camel:marshal ref="archiveFileInterfaceMetadataMapper"/>
                <camel:to pattern="InOnly"
                          uri="activemq:{{ccs.activemq.queue.prefix}}.sr.archive.bulk.ingestion.req?jmsMessageType=Text"/>
            </camel:split>
            <camel:log loggingLevel="INFO" message="Archiving content was processed"/>
        </camel:when>
        <camel:otherwise>
            <camel:log loggingLevel="INFO" message="No archiving content to process"/>
        </camel:otherwise>
    </camel:choice>
    </camel:transacted>
</camel:route>

路线 2：

<camel:route id="statementsArchivingBulkIngestionRequestRoute">
    <camel:from uri="activemq:{{ccs.activemq.queue.prefix}}.sr.archive.bulk.ingestion.req"/>
    <camel:throttle timePeriodMillis="4000">
    <camel:constant>1</camel:constant>
    <camel:setExchangePattern pattern="InOnly"/>
    <camel:unmarshal ref="archiveFileInterfaceMetadataMapper"/>
    <camel:bean ref="archiveFileEntryTransformer" method="transform"/>
    <camel:setHeader headerName="CamelHttpMethod">
        <camel:constant>POST</camel:constant>
    </camel:setHeader>
    <camel:toD uri="{{ccs.bulk.ingestion.service.ingest.archive.file}}"/>
    </camel:throttle>
</camel:route>

第一个路由中的处理器返回一个请求对象列表。然后拆分该列表，并将每个请求编组并放置在队列中。

第二条路由监听这个队列。当它使消息出队时，它会解组它，执行转换，然后使用它向另一个服务发送发布请求。我正在限制这条路由，使其每秒只处理一条消息，以免压垮下游服务。

当列表只包含几个请求时，这一切都很好，因此只有几条消息进入队列，但是当列表中有很多项目时，路由 2 超时并出现以下日志条目：

Atomikos:12] c.a.icatch.imp.ActiveStateHandler        : Timeout/setRollbackOnly of ACTIVE coordinator !

超时会导致该过程自行重复，下游服务最终会被每条消息多次调用，而不是仅调用一次。

我不明白为什么调用路由 2 的次数会导致它超时。我认为将为从 activemq 出队的每条消息启动一个路由实例。如果一条消息需要很长时间才能完成，那么我会理解，但显然超时是基于所有消息出队的累积时间。

我对 Camel 还很陌生，从架构的角度来看，我显然误解了一些东西。对于如何阻止这些超时发生的任何指导，我将非常感激。感谢您的阅读。

Answer 1

乔恩，你可能想尝试关注调查

1. 禁用/评论第二条路线。使用 activemq 的目的是使进程异步，这意味着第一条路线不应该因为第二条路线而产生任何影响。如果这在没有路线 2 的情况下有效，那么问题就出在其他地方。

2. 如果您发现第 1 条路线在没有第 2 条路线的情况下工作正常，那么接下来将尝试在第 2 条路线中设置更少的线程数。可以放置 1 或 2 个线程，看看是否有帮助。我认为这是对 activemq 而不是这些路由配置的争用。

3. 检查您推送到 activemq 的负载大小。如果您要向 activemq 发布非常大的消息，这也可能会产生影响，因为大量的项目并且每个项目都很大导致 activemq 中的争用并且事务花费的时间超过了超时设置。

如果您将大型数据集推送到 activemq，您可能需要重新审视设计。将有效负载保存在某个持久性区域（db/file/cache）中，并发送仅包含对有效负载和一些元数据的引用的通知事件。然后，第二条路由可以从事件中获取引用并检索路由 1 保存的有效负载。