java - Camel 文件序列号

Question

我原以为这是一个相当常见的用例，但我在 Camel 文档中没有看到任何内容。我查看了重新排序器，但似乎不合适。

1. 我有一个从第 3 方接收文件的路径。文件名包含序列号。文件必须按顺序处理。不能跳过任何序列号。
2. 如果收到的文件 > 比预期的序列号大，那么我想跳过它并等待正确的序列号。然后可以在收到预期的序列号后对其进行重新处理。
3. 如果接收到的文件 < 不是预期的序列，那么这将是一个错误。

用于此的最佳技术是什么？

我显然需要跟踪上次处理的/下一个预期的序列号。您将如何做到这一点，以便可以停止路线等，并保留对预期顺序的了解？

提前致谢。

Answer 1

您可以使用重新排序器(加上处理器)来执行此操作。

首先，要记住一些事情:从 camel 2.11 开始，重排序器可以配置为拒绝“旧”消息，但它不会(据我所知)检测“跳过”消息(例如消息 2 ,3,5 - 跳过 4)，因此您需要通过自定义处理器为此添加额外检查。

from("direct:myFiles")
        .bean(SetSequenceNumber.class)  // set header "seqnum" of type Long
        .resequence(header("seqnum")).stream().timeout(5000).rejectOld()
        .process(new Processor() {
            Long lastSeq;
            @Override
            public void process(Exchange exchange) throws Exception {
                Long seqnum = exchange.getIn().getHeader("seqnum", Long.class);
                if (lastSeq == null) {
                    lastSeq = seqnum;
                } else {
                    if (seqnum != lastSeq + 1) {
                        throw new IllegalStateException("Skipped sequence number between " + lastSeq + " and " + seqnum);
                    }
                    lastSeq = seqnum;
                }
            }
        })
        .log("Resequenced: ${header.seqnum}")
;

基本上，您可以从文件名中提取序列号并将其设置为名为 seqnum 的 header 。然后你通过重新排序器发送它，指定一个超时。然后它通过您的处理器检测到跳过的序列号。

希望对您有所帮助。