Apache骆驼路线中的Activemq并发失败 [英] Activemq concurrency fail in Apache camel route

问题描述

尝试在同一时刻向camel activemq 路由发送多个请求，一个请求得到服务，而另一个请求没有得到服务并按原样发回.Jms 消息在发送之前也设置了 JMScorrelationId

Trying to send multiple requests at same instant to camel activemq route, one request is serviced and the other request is not serviced and sent back as it is. The Jms messages are set with JMScorrelationId too before sending like below

textMessage.setJMSCorrelationID(UUID.randomUUID().toString());

下面是我的activemq路由

below is my activemq route

from("activemq:queue:TEST_QUEUE?disableReplyTo=true")
                .setExchangePattern(ExchangePattern.InOut)
                .process(new Processor() {
                    public void process(Exchange e) throws Exception {
                        log.info("Request : "
                                + MessageHelper.extractBodyAsString(e.getIn()));
                        /*Processing Logic*/
                    }
                })
                .beanRef("testBean","postDetails")
                .inOnly("activemq:queue:TEST_QUEUE");

同时发送到上述路由的多个(测试 2 个请求)请求没有得到服务，只有一个请求除外.servicemix.log 显示所有收到的请求.但只有一个得到服务.

Multiple (Test for 2 requests) requests sent to the above route concurrently not serviced except one. The servicemix.log shows all recieved requests. But only one is serviced.

以下是部署在 jboss 6.1 中作为 Web 应用程序的一部分发送请求的代码.

Below is the code what is sending request deployed in jboss 6.1 as part of web application.

public Message receive(String message, String queueName) {
        ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(
                "tcp://localhost:61616");
        String userName = "smx";
        String password = "smx";
        Connection connection;
        Message response =null;
        try {
            connection = connectionFactory.createConnection(userName, password);
            connection.start();
            ((ActiveMQConnectionFactory) connectionFactory)
                    .setDispatchAsync(false);
            Session session = connection.createSession(false,
                    Session.AUTO_ACKNOWLEDGE);
            Queue destination = session.createQueue(queueName);
            MessageProducer producer = session.createProducer(destination);
            producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
            TextMessage textMessage = session.createTextMessage(message);
            Queue tempQueue = session.createQueue(queueName);
            textMessage.setJMSReplyTo(tempQueue);
            producer.send(textMessage);
            MessageConsumer consumer = session.createConsumer(tempQueue);
            response = consumer.receive();
            response.acknowledge();

            session.close();
            connection.close();
        } catch (JMSException e) {
            e.printStackTrace();
        }
        return response;
    }

是否缺少某些或其他参数?请提出建议.

Is there some or the other parameter im missing?? please suggest.

推荐答案

如果 JMS 消息具有 JMSReplyTo 标头，Camel 将自动发回回复，因此您的路由应该是

Camel will auto send back a reply if the JMS message has a JMSReplyTo header, so your route should just be

from("activemq:queue:TEST_QUEUE")
                .process(new Processor() {
                    public void process(Exchange e) throws Exception {
                        log.info("Request : "
                                + MessageHelper.extractBodyAsString(e.getIn()));
                        /*Processing Logic*/
                    }
                })
                .beanRef("testBean","postDetails");

在路由的末尾(例如在调用 testBean 之后)，消息体的内容被用作回复消息，被发送回在 JMSReplyTo 标头中定义的队列.

At the end of the route (eg after calling testBean) then the content of the message body is used as the reply message, that are sent back to the queue named defined in the JMSReplyTo header.

这篇关于Apache骆驼路线中的Activemq并发失败的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！