标签: apache-camel

我正在尝试构建一个具有多个camel路由的应用程序,这些路由在内部重用了许多常用路由.因此,我试图在几个不同的Route Builder类中隔离路由,然后在需要的地方连接路由.

例如,所有与发送电子邮件有关的路由都进入EmailRouteBuilder类,所有处理特定JMS队列的路由都进入MyQueueRouteBuilder类.我认为这应该没问题,因为Camel不区分类,只查找路由定义.

另外,我还将几个异常处理路由分组到一个单独的ExceptionHandlingRouteBuilder中.

我也通过在Spring中定义camel context来连接所有不同的类,如此 -

<camelContext id="camelContext" xmlns="http://camel.apache.org/schema/spring">
    <propertyPlaceholder id="properties" location="classpath:${env}/autoimport.properties"/>
    <!-- Common Routes -->
    <routeBuilder ref="emailRouteBuilder" />
    <routeBuilder ref="myQueueRouteBuilder" />
  <routeBuilder ref="httpRouteBuilder" />
    <routeBuilder ref="exceptionsRouteBuilder" />
    <routeBuilder ref="customer1RouteBuilder" />
    <routeBuilder ref="customer2RouteBuilder" />

</camelContext>

我的例外路由器包含许多例外条款,如 -

onException(ConnectException.class)
            .routeId("connectExceptionEP")
            .handled(true)
            .log("Caught Exception: ")
            .to("direct:gracefulExit");

..
..
..

但是,看起来在另一个类中定义的异常存在问题,或者就此而言,在主路由定义中单独定义.

我通过查找正在引导的路由(通过routeId)并在检查何时抛出异常时在日志中验证了这一点.

另外,为了进一步确认,我采用了http Connect异常处理路由并将其直接放在httpRouteBuilder中并且... ..!,异常处理现在可以解决这个异常.

我在这里错过了一些东西,以便在其自己的类中很好地定义所有异常.？

我正在使用Apache Camel 2.9.0,但我在2.8.3中也验证了相同的行为.

谢谢,阿南德

我们使用Apache Camel作为编排引擎.通常,以下情形:

客户端发送HTTP请求< - > CAMEL代码< - >外部服务器

当我们的客户端向我们的CAMEL代码发送HTTP请求时,球开始滚动.Camel代码将通过REST HTTP调用触发外部服务器.最终,Camel代码会将回复发送回客户端.

在将响应发送回客户端之前的最后一个操作,Camel代码向外部服务器发送HTTP GET.因此首先设置TCP连接,然后发送数据.一段时间后(这可能需要5到10秒),外部服务器回复200 OK.

问题:收到200 OK后,Camel不会向外部服务器发送TCP FIN.结果,TCP连接保持打开状态...(外部服务器然后在超时200秒后关闭TCP连接,但这意味着TCP资源在200秒内丢失).

所以,在TCP级别,它是这样的:

Camel <---------->外部服务器

   TCP SYN  -->
   <-- TCP SYN,ACK
   TCP ACK  -->

   HTTP GET -->
   <-- 200 OK
   TCP ACK  -->

   <200 seconds later>
   <-- TCP FIN,ACK
   TCP ACK  -->

知道如何让Camel收到200 OK之后关闭TCP连接吗？

注意:我尝试添加"Connection:close"标题,但是Camel没有添加标题？!它似乎忽略了它......

这是添加标头的代码:

exchange.getOut().setHeader("Connection","Close");

我在使用Eclipse IDE的Spring框架中使用Camel 2.9.1.

如何在spring上下文中配置多个远程activemq代理(不同的IP地址)？以下是1个远程代理的配置.我使用camel创建路由,从多个远程代理中的不同队列生成和使用消息.根据以下路由,系统如何知道每个队列属于哪个远程代理？

• 项目清单

  从( "直接:启动")来.( "ActiveMQ的:队列:outgoingRequests")

• 项目清单

  from("activemq:queue:incomingOrders").to("log:Events？showAll = true").to("bean:jmsService")

1个代理org.camel.routes的Spring上下文

<bean id="jmsConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
    <property name="brokerURL" value="tcp://10.1.11.97:61616" />
</bean>

<bean id="pooledConnectionFactory"
    class="org.apache.activemq.pool.PooledConnectionFactory" init-
            method="start" destroy-method="stop">
    <property name="maxConnections" value="8" />
    <property name="connectionFactory" ref="jmsConnectionFactory" />
</bean>

<bean id="jmsConfig" class="org.apache.camel.component.jms.JmsConfiguration">
    <property name="connectionFactory" ref="pooledConnectionFactory"/>
    <property name="concurrentConsumers" value="10"/>
</bean>

<bean id="activemq" class="org.apache.activemq.camel.component.ActiveMQComponent">
    <property name="configuration" ref="jmsConfig"/>
</bean>

我正在使用Apache Camel根据请求消息中的某个属性路由SOAP请求.该消息被针对正则表达式匹配,并且如果找到匹配所述请求将被路由到"calldestination1",如果没有,它将被路由到"calldestination2".

我正在使用以下配置:

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:camel="http://camel.apache.org/schema/spring"
    xmlns:cxf="http://camel.apache.org/schema/cxf"
    xmlns:context="http://www.springframework.org/schema/context"
    xsi:schemaLocation="
    http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
    http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context.xsd
    http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd
    http://camel.apache.org/schema/cxf http://camel.apache.org/schema/cxf/camel-cxf.xsd">

<!-- ... -->

<cxf:cxfEndpoint id="testEndpointTest"
    address="http://localhost:8080/testEndpoint"
    endpointName="s:testEndpoint_Port"
    serviceName="s:testEndpoint"
    wsdlURL="wsdl/testEndpoint.wsdl"
    xmlns:s="http://teste.com/testEndpoint"/>

<!-- ... -->

<camelContext xmlns="http://camel.apache.org/schema/spring">

    <endpoint id="calldestination1" uri="http://localhost:8080/destination1?bridgeEndpoint=true&amp;throwExceptionOnFailure=false"/>
    <endpoint id="calldestination2" uri="http://localhost:8080/destination2?bridgeEndpoint=true&amp;throwExceptionOnFailure=false"/>

    <route streamCache="true">
        <!--CXF consumer using MESSAGE format--> 
        <from uri="cxf:bean:testEndpointTest?dataFormat=MESSAGE"/>

        <choice>
            <when>
                <simple>${bodyAs(java.lang.String)} regex ${properties:router.regex}</simple>
                <to uri="calldestination1"/>
            </when>
            <otherwise>
                <to uri="calldestination2"/>
            </otherwise>
        </choice>

    </route>

</camelContext>

当运行"calldestination2"的目标服务器处于加载状态时,请求可能需要大约1150ms才能响应.Apache Camel似乎没有很好地处理这个问题.

为了复制这种行为,我使用带有延迟(OnRequest Script)和jmeter的SOAP MockService的SoapUI.拳头我没有延迟地对SoapUI MockService进行测试,然后延迟1100ms.
然后我配置Apache Camel将请求路由到SoapUI服务并重复测试.

JMeter - > SoapUI - 0ms延迟
〜每秒1200个请求; 25ms请求平均值; 0%错误 …

我是阿帕奇骆驼的新手。我正在尝试将交换从 java 方法发送到路由，但它给出了“由 org.apache.camel.component.direct.DirectConsumerNotAvailableException：端点上没有可用消费者”错误。我想了解这个错误到底是什么以及我们什么时候收到这个错误？

@EndpointInject(uri = "direct:reportRoute")
private ProducerTemplate templatereportRoute;


public void saveDataFromExchange(Map<String, Object> DataMap){

    List<Map<String, Object>> paramList = new ArrayList<Map<String, Object>>();
    
    List<Map<String, Object>> rows = templatereportRoute.requestBody("direct:reportReport", DataMap, List.class);

<from uri="direct:reportRoute"/>

 <log message="  - ${body}"  loggingLevel="INFO"/>    

<setProperty propertyName="DataMap">
  <simple>${body}</simple>
</setProperty>

我遇到了一个问题，即执行器探测 JMS 运行状况失败，即使我的路由可以连接并向 JMS 生成消息。简而言之，执行器说它已关闭，但它正在工作。

技术堆栈和技术说明：

• Spring启动：2.3.1.RELEASE
• 骆驼：3.4.1
• 阿耳忒弥斯：2.11.0
• Artemis 已设置为使用用户名和密码(artemis/artemis)。
• 用于 org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory连接工厂。

我的路线就像筹码一样简单：

  <route id="timer-cluster-producer-route">
            <from uri="timer:producer-ticker?delay=5000"/>
          
            <setBody>
                <groovy>
                    result = ["Name":"Johnny"]
                </groovy>
            </setBody>
            <marshal>
                <json library="Jackson"/>
            </marshal>
            <to uri="ref:jms-producer-cluster-event" />
   </route>

基于 XML 的 Artemis 配置

由于 Spring-boot 支持基于 java 的配置，我正忙于相应地迁移我们的 XML beans。因此，我将一个工作 beans.xml 文件粘贴到项目中并启动了路由，我可以发送消息流，并且运行状况检查返回正常。

<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xsi:schemaLocation="
         http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
         http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd">
    <bean id="jmsConnectionFactory"
        class="org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616" />
        <property name="user" value="artemis"/>
        <property name="password" value="artemis"/>
        <property name="connectionLoadBalancingPolicyClassName" value="org.apache.activemq.artemis.api.core.client.loadbalance.RoundRobinConnectionLoadBalancingPolicy"/>
    </bean>
    <!--org.messaginghub.pooled.jms.JmsPoolConnectionFactory-->
    <!--org.apache.activemq.jms.pool.PooledConnectionFactory-->
    <bean id="jmsPooledConnectionFactory"
        class="org.apache.activemq.jms.pool.PooledConnectionFactory"
        init-method="start" destroy-method="stop"> …

我正在尝试使用Camel处理潜在的大文件,并且担心它们在Camel的主体中"适合" Message.有没有办法可以将文件的名称(路径)作为消息正文传递,然后在处理器中使用它来从磁盘读取？

我有一个要求,我需要每天重新加载我的osgi捆绑4次.重新加载bundle意味着重新创建静态实例Bean,重新加载camel路由,重新创建和注入线程池,数据库连接池..etc(其他spring xml东西).我尝试通过ssh刷新我的包,但我需要捆绑ID,以便可以改变加班.所以,我写了一个Manager Bundle,它通过符号名称获取捆绑包,并每天刷新它们4次

          osgi impl : felix 

          container : apache-servicemix-4.4.1-fuse-03-06

          Service Dependency spec : Blueprint

有3个包和一个帮助包.帮助包包含所有使用的公共类和服务接口.这三个捆绑包之间没有代码共享(它们都没有导出任何包).所有这些捆绑包都通过camel vm端点和服务进行交互.我只刷新其他3个捆绑包,帮助捆绑包不提供任何服务.

现在,问题是当我对这3个捆绑包进行更新时,它们启动并正常工作,但我看到每次执行此操作时jconsole上增加了800-900个类.强制gc似乎也没有清理这些对象.那么,这些旧的有线物体会是什么呢？服务依赖项应该自动更新,并且bundle之间没有代码依赖关系.我检查了更新之前和之前的类数量的差异.

我可以看到一些类的数量增加了一倍,如org.apache.activemq.camel.component.VmComponent, org.apache.commons.dbcp.BasicDataSource ..etc和我在我的camel路由中定义的一些自定义bean.我依赖于camel-core,blueprint,quartz等容器.在camel-context中使用bean,VM端点等等,以及在更新时在blueprint-config xml中定义的组件.我知道一旦更新捆绑包,建议调用FrameworkWiring.refreshBundles().但是,我没有捆绑之间的代码共享,我推测任何其他依赖容器应该处理我认为现在是错误的.我不确定servicemix中当前的felix框架实现是否支持FrameworkWiring.refreshBundles()(ref),我无法让它工作.我该如何解决这个问题？

谢谢你好

我们正在使用Apache Camel(Camel 2.10.3,基于Java DSL)构建集成项目.

我们有一个从数据库中提取数据的路径(让我们称之为IN_DB),做一些逻辑并每天插入另一个数据库(OUT_DB),另一个订阅XML数据的JMS主题的路径,做一些逻辑和插入它整天都在同一个数据库(OUT_DB)中.

要求是当JMS主题连接因任何原因而失效时,我们会不断尝试重新连接,一旦重新连接成功,我们需要返回数据库(IN_DB)并进行另一次加载以填补主题的空白失意了.

我的问题是我们怎么能在Camel中做这个逻辑('我已经连接然后我断开连接,现在又连接了')？当主题发生故障时,以主题消费者开头的路线会发生什么,路线会停止吗？或者它会向某个错误队列发出错误消息？我是否必须编写自己的处理程序来监视主题连接,或者当主题重新启动并设置一些消息头时Camel会自动重新连接,或者设置一些上下文变量以指示'我已连接然后我断开了连接现在我再次联系'情景已经发生了？我很高兴围绕调用数据库负载构建路由逻辑我无法找出在Camel中"检测"这种情况发生的最佳方法.

任何建议非常感谢.

我刚开始学习Apache Camel.我理解了路由和组件的基础知识.现在我想通过连接到Oracle数据库,从一个特定的表读取记录并使用File组件将这些记录写入文件来尝试.要从数据库中读取,我假设我需要使用JDBC组件并给出dataSourceName.

但是,我找不到有关如何使用camel创建dataSource的任何信息.我发现与此主题相关的所有信息都使用Spring DSL示例.我不使用Spring,我只需要使用简单的独立Java应用程序来测试它.

我正在使用JDK7u25和Apache Camel 2.12.1.

有人可以发布样本从oracle表中读取并写入文件吗？

[编辑]

在网上查看了几个解决方案之后,我开始了解以下两种方法:

1. 骆驼独立运行.这是我的代码:

   import javax.sql.DataSource;    
   import org.apache.camel.main.Main;
   import org.apache.camel.builder.RouteBuilder;
   import org.apache.commons.dbcp.BasicDataSource;
   
   public class JDBCExample {
   
       private Main main;
   
       public static void main(String[] args) throws Exception {
           JDBCExample example = new JDBCExample();
           example.boot();
       }
   
       public void boot() throws Exception {
           // create a Main instance
           main = new Main();
           // enable hangup support so you can press ctrl + c to terminate the JVM
           main.enableHangupSupport(); …

   Run Code Online (Sandbox Code Playgroud)