0

我正在使用 Apache camel 在 ActiveMQ 和 camel HTTP 端点之间进行路由。路由的定义方式是将数据从 camel-cxf webservice 获取到 ActiveMQ，并将这些数据发布到作为 HTTP 端点的 tomcat，然后一旦 http uri 命中，camel 从 tomcat 服务器获取响应。

我们能否仅在路由成功后从骆驼中确认队列，我的意思是在LOG_log_4过程完成后确认 AMQ？如果这是可能的，那么即使 http 链接断开，我也可以在队列中不丢失任何数据的情况下创建此路由。

我找到了这个链接JMS ACKNOWLEDGE_MODE。如果我需要使用确认模式 CLIENT_ACKNOWLEDGE 那么我是否需要在 tomcat Web 应用程序站点上编写消费者服务以确认收到的消息成功。

我使用 Acknowledgemodename 作为 AUTO_ACKNOWLEDGE。

详细路线图如下：-

在此处输入图像描述

Spring-DSL application-context.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:camel="http://camel.apache.org/schema/spring"
    xmlns:cxf="http://camel.apache.org/schema/cxf"
    xmlns:jaxrs="http://cxf.apache.org/jaxrs"
    xmlns:osgi="http://www.springframework.org/schema/osgi"
    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://www.springframework.org/schema/osgi       http://www.springframework.org/schema/osgi/spring-osgi.xsd       http://camel.apache.org/schema/spring       http://camel.apache.org/schema/spring/camel-spring.xsd       http://cxf.apache.org/jaxrs       http://cxf.apache.org/schemas/jaxrs.xsd       http://camel.apache.org/schema/cxf       http://camel.apache.org/schema/cxf/camel-cxf.xsd">
    <cxf:rsServer address="http://localhost:9000/employeeservice"
        id="restService" serviceClass="com.javainuse.beans.EmployeeServiceResource"/>
    <bean class="com.javainuse.beans.CamelProcessor" id="processor"/>
    <bean class="com.javainuse.beans.CamelAMQResponseProcessor" id="camelAMQResponseProcessor"/>
    <camelContext id="camelId" xmlns="http://camel.apache.org/schema/spring">
        <camel:route id="_route1">
            <camel:from id="_fromrest_1" uri="cxfrs://bean://restService"/>
            <camel:process id="_processrest_1" ref="processor"/>
            <camel:to id="_to1" uri="amq:queue:queue1x"/>
            <camel:log id="_log1" message="hit ws route 1"/>
        </camel:route>
        <camel:route id="_route2">
            <camel:from id="_from2" uri="amq:queue:queue1x?concurrentConsumers=2"/>
            <camel:log id="_log2" message="hit amq route 2"/>
            <camel:log id="_log3" message="Message returned from the queue is ${body}"/>
            <camel:setBody id="_setBody2">
                <simple>${body}</simple>
            </camel:setBody>
            <setHeader headerName="CamelHttpMethod" id="_setHeader1">
                <constant>PUT</constant>
            </setHeader>
            <setHeader headerName="Cookie" id="_setHeader2">
                <constant>JSESSIONID=Z2rLMNdkmpz72nG2FTj5SuLHn_rbZxHzs7wbw3LG</constant>
            </setHeader>
            <setHeader headerName="Authorization" id="_setHeader3">
                <constant>eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiJQQVNTV09SRC9hcHBhZG1pbiIsImF1dGgiOiJERUYtcm9sZS1hcHBhZG1pbiIsImV4cCI6MTUzNTQzNjg3Mn0.AqfYVoAFsh6XDiIGGVO8lRpGyb1HcdP2HN1rUWtfDbYNJ6rdsnLnT5nrqmtPSR7YfUui7pWKxEF96NcLKppQAg</constant>
            </setHeader>
            <camel:to id="_to2" uri="http:localhost:8080/api/outboundmsg/"/>
            <camel:log id="_log4" message="Message returned from iConnect is ${body} with header ${in.headers.CamelHttpResponseCode}"/>
        </camel:route>
    </camelContext>
</beans>

ActiveMQ 配置 applicationContext-mq.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:camel="http://camel.apache.org/schema/spring"
    xmlns:cxf="http://camel.apache.org/schema/cxf"
    xmlns:osgi="http://www.springframework.org/schema/osgi"
    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://www.springframework.org/schema/osgi http://www.springframework.org/schema/osgi/spring-osgi.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-2.8.3.xsd">
    <bean class="org.apache.activemq.camel.component.ActiveMQComponent" id="amq">
        <property name="connectionFactory" ref="singleCF"/>
        <property name="useSingleConnection" value="true"/>
        <property name="usePooledConnection" value="false"/>
        <property name="preserveMessageQos" value="true"/>
        <property name="acknowledgementModeName" value="AUTO_ACKNOWLEDGE"></property>
    </bean>
    <bean
        class="org.springframework.jms.connection.SingleConnectionFactory" id="singleCF">
        <property name="targetConnectionFactory" ref="AMQCF"/>
    </bean>
    <bean class="org.apache.activemq.ActiveMQConnectionFactory" id="AMQCF">
        <property name="userName" value="admin"/>
        <property name="password" value="admin"/>
        <property name="brokerURL" value="tcp://localhost: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="AMQCF" />
    </bean>
    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="pooledConnectionFactory" />
    </bean>
</beans>

先感谢您。

4

1 回答 1

0

代理在消费者消费后立即收到确认。即在你的情况下<camel:from id="_from2" uri="amq:queue:queue1x?concurrentConsumers=2"/>

您的要求尚不清楚，但是如果您需要通过队列实现请求回复，您可以将replyTo 参数设置为队列并将从tomcat 接收到的响应推送到该replyTo 队列。或者，让我知道确切的要求。

于 2018-08-29T15:06:41.363 回答