Вопрос по java, concurrency, activemq, apache-camel – Apache Camel: Ответ получен для неизвестного идентификатора

1

Eстьmiddleware между двумя другими программными компонентами. В яmiddlewareм маршрутизацииApache ActiveMQ сообщения от.Apache Camel

вот как это работает:

1stComponent использованияmiddleware отправить сообщение3rdComponent

3rdComponent отвечает на сообщение и отправляет его обратно1st(с помощьюmiddleware).

           1stComponent  Middleware  3rdComponent

Проблема: Я

м используюConcurrentConsumers в промежуточном программном обеспечении.

В середине отправки много сообщений последовательно, внезапноmiddleware останавливает весь процесс! нет никаких исключений или сообщений! например, обработаны первые 100 из 500 сообщений, а оставшиеся остаются в очереди как ожидающие сообщения.

это предупреждение записывается иногда в середине процесса:

[WARN ] TemporaryQueueReplyManager(Camel (camel-1) thread #11 - TemporaryQueueReplyManager[Q.MyQ]):91 - Reply received for unknown correlationID [c551c7aa061f501c]. The message will be ignored: ActiveMQMapMessage {commandId = 2161, responseRequired = true, messageId = ID:xxxxxxx, originalDestination = null, originalTransactionId = null, producerId = ID:xxxxxxx, destination = temp-queue://ID:localhost.localdomain-40961-1389890357282-3:1:1, transactionId = null, expiration = 0, timestamp = 1389890272360, arrival = 0, brokerInTime = 1389890272360, brokerOutTime = 1389890272360, correlationId = c551c7aa061f501c, replyTo = temp-queue://ID:localhost.localdomain-40961-1389890357282-3:1:1, persistent = true, type = null, priority = 4, groupID = null, groupSequence = 0, targetConsumerId = null, compressed = false, userID = null, content = null, marshalledProperties = [email protected], dataStructure = null, redeliveryCounter = 0, size = 0, properties = {breadcrumbId=ID:xxxxxxxxxxxxxx, Title=300, CamelJmsDeliveryMode=1}, readOnlyProperties = true, readOnlyBody = true, droppable = false, jmsXGroupFirstForConsumer = false} ActiveMQMapMessage{ theTable = {} }

этоmiddlewares Код:

private static class MyRouteBuilder extends RouteBuilder {
    @Override
    public void configure() throws Exception {
        from("activemq:queue:Q.Middleware?concurrentConsumers=1&maxConcurrentConsumers=10")
        .threads(1, 100)
            .process(new Processor() {
                public void process(Exchange exchange) {
                    //some code
                }
            })
        .inOut("activemq2:queue:Q.3RD")
        ;
    }
}

и3rdComponent

private static class MyRouteBuilder extends RouteBuilder {
    @Override
    public void configure() {
        from("activemq:queue:Q.3RD")
        .threads(1, 100)
        .process(new Processor() {
            public void process(Exchange exchange) {
                //some code
            }
        })
        ;
    }
}
@Ralf: пожалуйста, смотрите мой комментарий для Dinesh 'ответ Amin Sh
Кажется, все ваши темы либо заняты, либо ждут ответов. Я думаю, что сообщение об ошибке, которое вы получаете, означает, что Camel просрочил запрос, потому что TTL был превышен. Но потом все-таки получил ответ. Видите ли вы другие ошибки, указывающие на это? Проверять, выписыватьсяО времени жить в доке Camel-JMS Ralf

Ваш Ответ

4   ответа
1

использовать парадигму запрос / ответ (которую выиспользуя InOut в своем маршруте), производитель ожидает получить ответы на каждое сообщение, которое он отправляет, прежде чем отправит следующее. Если два производителя выдают сообщения, которые придут к одному и тому же месту назначения ответа, и ониВы смотрите этот пункт назначения с тем же селектором, а затемs условие гонки, приходит ли сообщение к производителю, что 'ожидает этого или другого, и если это идет к другому, вы 'я получу сообщениеВы увидите в источнике, который получил неожиданный ответ, и сообщение, указывающее время ожидания ожидания ответа в источнике, который отправил исходное сообщение.

Тот'почему твоиответ" (который нена самом деле отвечает на ваш вопрос, только опишите, как надежно воспроизвести проблему) способен воспроизвести проблему, запустив несколько параллельных маршрутов, которые отправляют сообщения InOut в один и тот же пункт назначения.

Чтобы решить эту проблему, необходимо убедиться, что ваши производители будут получать только свои собственные сообщения, либо через отдельные пункты назначения ответов для каждого производителя, либо с помощью уникальных селекторов для каждого производителя, либо путем сокращения числа производителей.

-1

Это'Можно симулировать эту функцию, используя несколько маршрутов, как это:

...
        // ----<1st Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<2nd Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<3rd Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<4th Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<5th Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<6th Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<7th Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<8th Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<9th Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
        // ----<10th Route>----
        from("activemq:queue:Q.Middleware").threads(2, 20).inOut("activemq2:queue:Q.3RD");
...

Работает нормально, но справляетсяКоличество потребителей (Маршруты) не подходит! Вы должны скопировать & Вставить маршруты!

Вам следуетопределенно не нужно этого делать. Смотрите другие ответы Roy Truelove
0

@ Амин Ральф прав, добавляя больше к своему ответу - может быть две причины, по которым процесс остановлен 1 - TTL истек 2 - Время между клиентом и сервером не синхронизировано.

Если ваша проблема связана с 1, тогда установите заголовок - "JMSExpiration»

Если ваша проблема связана с 2 - (копия вставлена из другого сообщения stackoverflow)

тогда часы между клиентом и брокером должны быть синхронизированы, чтобы срок действия работал правильно. Если часы не синхронизированы, то срок действия, установленный от клиента, может уже истечь, когда сообщение получено посредником. Или время клиента опережает брокера, поэтому срок действия превышает 10 секунд.

Чтобы исправить это, выровняв время, будьте только на основе брокера. Увидетьhttp://activemq.apache.org/timestampplugin.html

@AminSh А ты тоже изменил requestTimeout? По умолчанию это 20 секунд. Если у вас нет TTL для сообщения MQ, и для получения ответа требуется более 20 секунд, то я ожидаю увидеть полученную ошибку Но тогда вы также должны увидеть тайм-аут где-то в своих логах. Ralf
спасибо но я неЯ так не думаю. Я отключилTTL но все равно ничего. предупреждение говорит о том, чтоReply received for unknown correlationID Amin Sh
@Ralf: Да, я сделал это. если тайм-аут случается, верблюд бросает исключение. Amin Sh
0

ОБНОВИТЬ:

Мой предыдущий ответ работает правильно, но не было полного решения.

Ошибка былапроизводстванеуникальным !CorrelationIDs (ошибка в генераторах случайных строк) просто !!! : |

Похожие вопросы