1

我现在在骆驼中做了一些小项目，但我很难理解的一件事是在骆驼路线中消费时如何处理大数据（不适合内存）。

我有一个数据库，其中包含我想使用骆驼路由的几 GB 的数据。显然，将所有数据读入内存不是一种选择。

如果我将其作为一个独立的应用程序来执行，我将拥有对数据进行分页并将块发送到我的 JMS 端点的代码。我想使用骆驼，因为它提供了一个很好的模式。如果我从文件中消费，我可以使用 streaming() 调用。

我还应该使用 camel-sql/camel-jdbc/camel-jpa 还是使用 bean 从我的数据库中读取数据。

希望大家还在我身边。我对 Java DSL 更熟悉，但如果有人能提供任何帮助/建议，我将不胜感激。

更新：2012 年 5 月 2 日

所以我有一些时间来解决这个问题，我认为我实际上在做的是滥用 Producer 的概念，以便我可以在路线中使用它。

public class MyCustomRouteBuilder extends RouteBuilder {

    public void configure(){
         from("timer:foo?period=60s").to("mycustomcomponent:TEST");

         from("direct:msg").process(new Processor() {
               public void process(Exchange ex) throws Exception{
                   System.out.println("Receiving value" : + ex.getIn().getBody() );
               }
         }
    }

}

我的制作人如下所示。为清楚起见，我没有包含 CustomEndpoint 或 CustomComponent，因为它似乎只是一个薄包装器。

public class MyCustomProducer extends DefaultProducer{ 

    Endpoint e;
    CamelContext c;

    public MyCustomProducer(Endpoint epoint){
          super(endpoint)   
          this.e = epoint;
          this.c = e.getCamelContext();
    }

    public void process(Exchange ex) throws Exceptions{

        Endpoint directEndpoint = c.getEndpoint("direct:msg");
        ProducerTemplate t = new DefaultProducerTemplate(c);

        // Simulate streaming operation / chunking of BIG data.
        for (int i=0; i <20 ; i++){
           t.start();
           String s ="Value " + i ;                  
           t.sendBody(directEndpoint, value)
           t.stop();         
        }
    }
}

首先，上面看起来不是很干净。执行此操作的最简洁方法似乎是通过计划的石英作业填充 jms 队列（代替直接：msg），然后我的骆驼路线会使用该作业，以便我可以更灵活地控制骆驼中收到的消息大小管道。但是，我非常喜欢将基于时间的激活设置为 Route 的一部分的语义。

有没有人对最好的方法有任何想法。

4

1 回答 1

1

据我了解，您需要做的就是：

from("jpa:SomeEntity" + 
    "?consumer.query=select e from SomeEntity e where e.processed = false" +
    "&maximumResults=150" +
    "&consumeDelete=false")
.to("jms:queue:entities");

maximumResults定义每次查询获得的实体数量的限制。

实体实例处理完毕后，需要设置它e.processed = true;，persist()使实体不再被处理。

一种方法是使用@Consumed注释：

class SomeEntity {
    @Consumed
    public void markAsProcessed() {
        setProcessed(true);
    }
}

另一件事，您需要注意的是如何在将实体发送到队列之前对其进行序列化。您可能需要在 from 和 to 之间使用浓缩器。

于 2012-04-24T12:21:23.823 回答