我想开发一个轮询包含CSV文件的目录的路由,并且对于每个文件,它使用Bindy解组每一行并在activemq中对其进行排队.
问题是文件可能非常大(一百万行)所以我更喜欢一次排队一行,但我得到的是Bindy末尾的java.util.ArrayList中的所有行记忆问题.
到目前为止,我有一点测试和解组工作,所以使用注释的Bindy配置是可以的.
这是路线:
from("file://data/inbox?noop=true&maxMessagesPerPoll=1&delay=5000")
.unmarshal()
.bindy(BindyType.Csv, "com.ess.myapp.core")
.to("jms:rawTraffic");
环境是:Eclipse Indigo,Maven 3.0.3,Camel 2.8.0
谢谢
该DirectComponent文档给出了下面的例子:
from("activemq:queue:order.in")
.to("bean:orderServer?method=validate")
.to("direct:processOrder");
from("direct:processOrder")
.to("bean:orderService?method=process")
.to("activemq:queue:order.out");
这和下面有什么区别吗？
from("activemq:queue:order.in")
.to("bean:orderServer?method=validate")
.to("bean:orderService?method=process")
.to("activemq:queue:order.out");
我试图找到关于to()方法在Java DSL上的行为的文档,但是除了RouteDefinition javadoc之外(它给出了非常简单的"将交换发送到给定的端点")我发现空白:(
我想监视给定目录中的所有文件以进行更改,即更新的时间戳.这个用例对于使用文件组件的Camel来说似乎很自然,但我似乎找不到配置此行为的方法.
一个uri喜欢:
file:/some/directory
将使用提供的目录中的文件,但将删除它们.
一个uri喜欢:
file:/some/directory?noop=true
在添加每个文件或路由启动时消耗每个文件一次.
令人惊讶的是,没有一个选项
consumeOnChange=true
是否有一种简单的方法来监控文件更改而不是在使用后删除文件？
我对 Akka的理解是它提供了一个模型,通过这种模型,多个隔离的线程可以以高度并发的方式相互通信.它使用"actor模型",其中每个线程都是一个具有特定工作的"actor".您可以协调哪些消息在什么条件下传递给哪些actor.
我之前使用过Camel,对我来说,我觉得它现在已经失去了它的光彩/效用,因为Akka是如此成熟和有据可查.据我了解,Camel是关于企业集成的,即将多个不同的系统集成在一起,通常以某种服务总线方式.
但想一想:如果我目前正在使用Camel:
我可以和Akka做同样的事情; 我可以为每个步骤(轮询FTP,转换文件 - > POJO,电子邮件或持久化)拥有1个Actor,将它们连接在一起,让Akka处理所有的异步/并发.
所以尽管Akka是一个并发框架(使用actor),尽管Camel是关于集成的,但我不得不问:Akka不能解决Camel所做的一切吗？换句话说:在Akka上使用Camel仍然存在哪些用例？
我想设置一个spring bean(通过接口或bean类).我可以打电话来"开始"一条路线.
在这个简单的例子中,当我从代码中调用sayHello("world")时,我希望将sayHello方法的返回值路由到将其写入文件的端点.
有谁知道这是否可行,或者如何解决这个问题？我知道我可以通过CXF公开相同的接口并使其工作,但我真的只想调用一个方法,而不是经历发送jms消息或调用webservice的麻烦.
public interface Hello{
public String sayHello(String value);
}
from("bean:helloBean").to("file:/data/outbox?fileName=hello.txt");
Camel显式处理错误处理的两个"范围":
我遇到的问题是在"子路线"中抛出的异常.例如,我有这条路线:
from("direct:sendToWebservice").
.processRef("massageBeforeSending").
.to("http://webservice.com").
.processRef("massageResponse");
然后我有两个需要向web服务发送消息的其他路由:
from(direct:fromSystemA").
.errorHandler(deadLetterChannel("direct:TellSystemA")).
.to("direct:sendToWebservice");
from(direct:fromSystemB").
.errorHandler(deadLetterChannel("direct:TellSystemB")).
.to("direct:sendToWebservice");
我想要发生的是,如果webservice路由抛出异常,它会传播到调用者,系统A或系统B将被通知.我没有看到实现这一目标的方法.
我觉得这将是一个常见的用例 - 有人曾经碰过它吗？
感谢你的宝贵时间,
罗伊
我已经编写了使用apache camel调用rest api的示例代码.哪个在独立工作正常,但是我用来创建OSGI捆绑包的相同代码并将其部署到karaf容器中,捆绑包是成功创建的但我得到的错误如"没有组件找到方案http"当我尝试叫它.
你能帮我解决这个问题吗？
这是代码:
CamelContext context = new DefaultCamelContext();
context.addRoutes(new RouteBuilder() {
public void configure() {
from("direct:start")
.setHeader(Exchange.HTTP_METHOD,simple("GET"))
.to("http://10.10.10.10:8080/RestfulDemo/rest/get");
}
});
context.start();
ProducerTemplate template = context.createProducerTemplate();
String headerValue = "application/xml";
Map<String, Object> headers = new HashMap<String,Object>();
headers.put("Content-Type", headerValue);
Object result = template.requestBodyAndHeaders("direct:start", null, headers, String.class);
Exchange exchange = new DefaultExchange(context);
String response = ExchangeHelper.convertToType(exchange, String.class, result);
System.out.println("Response : "+response);
context.stop();
错误如下:
org.apache.camel.ResolveEndpointFailedException: Failed to resolve endpoint: http://10.10.10.10:8080/RestfulDemo/rest/get due to: No component found with scheme: http
休息端点
<jaxrs:server id="jaxrs"
address="http://127.0.0.1:8080/jaxrs">
<jaxrs:serviceBeans>
<ref component-id="service1" />
...
...
<ref component-id="serviceX" />
</jaxrs:serviceBeans>
<jaxrs:providers>
<ref component-id="runtimeExceptionMapper" />
</jaxrs:providers>
</jaxrs:server>
路线
<route id="secureBridgeRoute">
<from uri="jetty:https://0.0.0.0:443/jaxrs?sslContextParametersRef=sslContextParameters&matchOnUriPrefix=true&minThreads=8&maxThreads=16" />
<transacted ref="JTA_TRANSACTION" />
<to uri="jetty:http://127.0.0.1:8080/jaxrs?bridgeEndpoint=true&throwExceptionOnFailure=true" />
</route>
DAO
<bean id="dao1" class="com.example.Dao1" activation="eager">
<jpa:context unitname="PU" property="entityManager" type="TRANSACTION" />
</bean>
服务bean
<bean id="service1" class="com.example.Service1" activation="eager">
<property name="dao1" ref="dao1" />
<property name="dao2" ref="dao2" />
<tx:transaction method="*" value="Required" />
</bean>
服务bean方法伪代码
boolean create(entity1, entity2) {
dao1.persist(entity1);
dao2.persist(entity2);
}
当dao2持久失败时,事务没有回滚.Entity1被插入到DB中.
附加信息
1)TransactionManager定义
<reference id="platformTransactionManager" interface="org.springframework.transaction.PlatformTransactionManager" />
<bean id="JTA_TRANSACTION" class="org.apache.camel.spring.spi.SpringTransactionPolicy"> …我试图拆分一个ArrayList并使用Apache Camel将每个元素写入自己的文件,就像在这个简化的例子中一样:
from("timer://poll?period=10000").process(new Processor(){
public void process(Exchange exchange){
ArrayList<String> list = new ArrayList<String>();
list.add("one");
list.add("two");
list.add("three");
exchange.getIn().setBody(list, ArrayList.class);
}
}).split(body()).log(body().toString()).to("file:some/dir");
日志打印每个项目,但只有"三"保存到文件.我究竟做错了什么？
一月