Apache Camel重新排序程序不工作

Question

我正在尝试使用Apache Camel运行一个简单的重新排序程序。该程序使用Java DSL对传入的Java消息重新排序。当我运行这个程序时，消息会被写入文件夹，但根据标题值或消息正文中单个单词的字母顺序，不会出现任何特定的顺序。Camel创建的文件仍然无序，就好像resequenceDSL函数什么都没做一样。

如何让这个程序像Arrays.sort（）方法那样对消息进行实际排序？此外，如何让这个程序重新排序，然后以正确的排序顺序将消息聚合到一个文件中？

这是节目。。。我通过另一个具有main方法的类调用主Camel route。

import org.apache.camel.builder.RouteBuilder;

public class SortThoseMessages extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        from("direct:pointA")
                .resequence(header("grocery"))

                .to("file:target/pointB");
    }

}

下面的类具有main，并将消息生成到队列pointA中。

import org.apache.camel.CamelContext;
import org.apache.camel.impl.DefaultCamelContext;
import org.apache.camel.ProducerTemplate;

public class NewSequenceMain {

    public static void main(String[] args) { 
        CamelContext c = new DefaultCamelContext();

        try { 

         c.addRoutes(new SortThoseMessages());



        ProducerTemplate template = c.createProducerTemplate();

        c.start();

        template.sendBodyAndHeader("direct:pointA", "apple", "grocery", 1);
        template.sendBodyAndHeader("direct:pointA", "orange", "grocery", 3);
        template.sendBodyAndHeader("direct:pointA", "bannanna", "grocery", 2);

        Thread.sleep(5000);
        c.stop();

        } catch(Exception ex) { 
            System.err.println("Exception thrown -> " + ex);
            System.err.println("Now printing stacktrace...");
            ex.printStackTrace();
        }

    }

}

Answer 1

消息可能会在驼峰路径内重新排序，但在写入文件时不会。要查看重新排序，请调用聚合器JavaDSL，以按指定的顺序查看消息正文。程序中发布的消息是根据标题的编号排序的。ProducerTemplate对象引用上的方法调用将头设置为sendboyandheader（）方法调用的最后一个参数中的整数。

要将重新排序作为骆驼路线的目的地在单个文件中生效，请查看以下示例：

import org.apache.camel.builder.RouteBuilder;


public class ReorganizingMessages extends RouteBuilder {

    @Override
    public void configure() throws Exception {

        from("direct:pointA")
                .resequence(header("grocery"))
                .to("log://org.apache.camel.howto?showAll=true&multiline=true")

                .aggregate().constant(true).completionTimeout(100L).
                aggregationStrategy(new StringAggregator())
                .to("file:target/pointB");
    }

}

上面的代码使用了一个自定义聚合器Javabean，如下所示。

import org.apache.camel.Exchange;
import org.apache.camel.processor.aggregate.AggregationStrategy;

public class StringAggregator implements AggregationStrategy {

    @Override
    public Exchange aggregate(Exchange old, Exchange new1) {
        if (old == null) { 
            return new1;
        }

        String oldBody = old.getIn().getBody(String.class);
        String newBody = new1.getIn().getBody(String.class);
        old.getIn().setBody(oldBody + " " + newBody);
        return old;

    }

}