Проблема с несколькими потребителями в activeMQ

У нас есть два экземпляра activemq, настроенные как кластер, и кластер из четырех экземпляров glassfish, которые должны быть подключены к ОДНОМ activemq в любой момент времени. Если activemq01 недоступен, все четыре экземпляра glassfish должны переключиться на activemq02.

Я несколько раз замечал, что по неизвестной причине один из (случайных) экземпляров Glassfish будет переключаться на activemq02, а оставшиеся три экземпляра будут по-прежнему подключены к activemq01, даже если activemq01 НЕ был отключен, и все экземпляры Glassfish должны были прослушивать на activemq01.

Журналы не указывают ничего, что могло бы объяснить это поведение, все, что я могу видеть, если этот экземпляр Glassfish не может подключиться к activemq01 и выполнить сбой на activemq02.

Были ли у кого-нибудь такие же или похожие проблемы? Любая помощь / совет приветствуются.

Вот мои конфиги activemq:

activemq01:

http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">

файл: $ {activemq.base} /conf/credentials.properties

<managementContext>
  <managementContext connectorPort="1093" createConnector="true"/>
</managementContext>

<networkConnectors>
  <networkConnector name="amq-prod" uri="static://(tcp://127.0.0.1:61616,tcp://192.168.0.167:61616)" />
</networkConnectors>


<persistenceAdapter>
  <kahaDB directory="${activemq.base}/data/kahadb"/>
</persistenceAdapter>

<plugins>
  <loggingBrokerPlugin logAll="false" logConnectionEvents="true"/>
</plugins>


<systemUsage>
  <systemUsage>
    <memoryUsage>
      <memoryUsage limit="2048 mb"/>
    </memoryUsage>
    <storeUsage>
      <storeUsage limit="2 gb" name="prod"/>
    </storeUsage>
    <tempUsage>
      <tempUsage limit="2000 mb"/>
    </tempUsage>
  </systemUsage>
</systemUsage>

<transportConnectors>
  <transportConnector name="openwire" uri="tcp://0.0.0.0:61616" updateClusterClients="true" />
</transportConnectors>

#

activeMQ02:

http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-2.0.xsd http://activemq.apache.org/schema/core http://activemq.apache.org/schema/core/activemq-core.xsd">

файл: $ {activemq.base} /conf/credentials.properties

<managementContext>
  <managementContext connectorPort="1093" createConnector="true"/>
</managementContext>

<networkConnectors>
  <networkConnector name="amq-prod" uri="static://(tcp://127.0.0.1:61616,tcp://192.168.0.166:61616)"  />
</networkConnectors>


<persistenceAdapter>
  <kahaDB directory="${activemq.base}/data/kahadb"/>
</persistenceAdapter>


<systemUsage>
  <systemUsage>
    <memoryUsage>
      <memoryUsage limit="2048 mb"/>
    </memoryUsage>
    <storeUsage>
      <storeUsage limit="2 gb" name="prod"/>
    </storeUsage>
    <tempUsage>
      <tempUsage limit="2000 mb"/>
    </tempUsage>
  </systemUsage>
</systemUsage>

<transportConnectors>
  <transportConnector name="openwire" uri="tcp://0.0.0.0:61616" updateClusterClients="true" />
</transportConnectors>

версия activeMQ - 5.4.1


activemq glassfish
person Nerses    schedule 14.04.2011    source источник

Ответы (1)


arrow_upward
1
arrow_downward

Определите свой тег networkConnector следующим образом:

Конфигурация acticemq01:

 <networkConnector
         name="amq1-nc" 
         uri="static:(failover:(tcp://192.168.0.167:61616))" 
         networkTTL="2" 
         duplex="true"
         dynamicOnly="true" 
  />

Конфигурация acticemq02:

 <networkConnector
         name="amq2-nc" 
         uri="static:(failover:(tcp://192.168.0.166:61616))" 
         networkTTL="2" 
         duplex="true"
         dynamicOnly="true" 
  />

А в ваших потребителях используйте URL-адрес поставщика JMS следующим образом:

failover:(tcp://192.168.0.166:61616,tcp://192.168.0.167:61616)?randomize=false&timeout=5000

Вышеуказанный URL-адрес failover всегда будет подключаться к брокеру acticemq01, если он доступен. Если acticemq01 выйдет из строя, он автоматически переключится на брокера acticemq02. Также timeout = 5000 гарантирует, что потребитель вернет ошибку через 5 секунд при попытке подключиться к любому из 2 брокеров.

person anubhava    schedule 14.04.2011
comment
Спасибо, Анубхава! Проблема с этой конфигурацией заключается в том, что, когда я останавливаю activemq01, соединения на activemq02 терпят неудачу, но когда я попытался снова запустить activemq01 и остановил activemq02, потребители не вернулись к activemq01. - person Nerses; 15.04.2011
comment
моя очередь ожидания продолжает накапливаться, и я потратил много времени на изучение и настройку параметров, но не уверен, что происходит. Это происходит ТОЛЬКО с одной из очередей. Просто чтобы дать вам немного информации, на случай, если вы можете посоветовать улучшить производительность с помощью некоторых изменений конфигурации: - person Nerses; 15.04.2011
comment
производитель помещает объекты сообщения в очередь, которая имеет несколько десятков отдельных контактов, затем потребители (4) прослушивают эту очередь и выбирают сообщение (объект), анализируют и доставляют поставщику удаления, по одному контакту за раз. Известно, что доставка поставщика медленная. Как вы думаете, если я увеличу размер предварительной выборки в этой очереди (в настоящее время 100), это исправит проблему, и сообщения не будут складываться в очереди ожидания? заранее спасибо - person Nerses; 15.04.2011
comment
Желательно, чтобы, если вы знаете, что конкретный потребитель будет медленным, установите его размер предварительной выборки на что-то меньшее. Чтобы узнать, как обращаться с большими ожидающими сообщениями, я бы посоветовал вам взглянуть на эту страницу: activemq. apache.org/slow-consumer-handling.html - person anubhava; 15.04.2011
comment
Вот процесс: одно сообщение в очереди может содержать 100 контактов, поэтому один из экземпляров Glassfish заберет сообщение из очереди, затем проанализирует сообщение объекта и отправит каждый контакт индивидуально, что занимает около 4 минут. Насколько я понимаю, поскольку activemq ожидает подтверждения раньше, чем через 4 минуты, он помечает их как повторную доставку и помещает в очередь ожидания. Есть ли способ увеличить это время в activemq для этой конкретной очереди, чтобы он ждал около 4 минут для acks / nacks? или даже не ждать подтверждения, просто выстрелить и забыть? - person Nerses; 15.04.2011
comment
ACK отправляется немедленно, как только сообщение получено потребителями, оно не задерживается для дальнейшей обработки на потребителе. Сколько параллельных потоков-потребителей вы создаете для каждого экземпляра-потребителя? Я бы создал 16 на экземпляр. Для моего контейнера слушателя сообщений Spring я использую для этого `‹ property name = concurrentConsumers value = 16 / ›`. Но для производителей вы, конечно, можете использовать jms.useAsyncSend=true в конце URL-адреса поставщика JMS, чтобы сделать отправку асинхронной. - person anubhava; 16.04.2011
comment
Спасибо, Анубхава. Я проверяю через jconsole, и максимальное количество одновременных подключений, которые я вижу для каждого экземпляра Glassfish, равно 6, однако каждый экземпляр Glassfish может достигать 32. Я пробовал использовать ‹property name = concurrentConsumers value = 16 /›, но это не так. t работать, activemq даже не запустится с этим свойством. - person Nerses; 19.04.2011
comment
‹? Xml version = 1.0?› ‹Beans xmlns = springframework.org/schema/beans xmlns : amq = activemq.apache.org/schema/core xmlns: xsi = w3.org/2001/XMLSchema-instance xsi: schemaLocation = springframework.org/schema/beans springframework.org/schema/beans/spring-beans-2.0.xsd activemq.apache.org/schema/core activemq.apache.org/schema/core/activemq-core.xsd› - person Nerses; 19.04.2011
comment
‹Bean class = org.springframework.beans.factory.config.PropertyPlaceholderConfigurer› ‹имя свойства = местоположения› ‹value› файл: $ {activemq.base} /conf/credentials.properties ‹/value› ‹/property› ‹имя свойства = concurrentConsumers value = 16 / ›‹/bean› - person Nerses; 19.04.2011
comment
К сожалению, <property name="concurrentConsumers" value="16" /> нельзя использовать в конфигурации AMQ. Он используется в контейнере прослушивателя сообщений Spring для установки количества одновременных прослушивателей для каждого экземпляра потребителя. Если ваши потребители работают очень медленно, занимают много времени, вы увидите ожидающие сообщения на брокере ActiveMQ, это ожидаемое поведение. например: если каждый экземпляр потребителя может достигать 32, тогда вы можете одновременно обрабатывать 32x3 = 128 сообщений, а поскольку каждое сообщение занимает до 4 минут, все поступающие сообщения переходят в состояние ожидания. - person anubhava; 19.04.2011