netty - MessageListener 停止监听 HornetQ 队列

问题描述

以下问题正在损害我的生产系统。所以我有多个 MDB 打包为 EAR、WAR 应用程序并部署在 JBOSS 中。当我的网站有大量流量时，这些 MDB 停止监听写入 HornetQ 队列的消息，我被迫重新启动系统。上次发生这种情况时，我编写了一个独立的消息监听器，并且能够监听来自同一个总部服务器的消息。这指出问题出在我的应用程序服务器/应用程序级别。我附上以下内容：

典型的 MDB

@MessageDriven(activationConfig = {
    @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"), 
    @ActivationConfigProperty(propertyName = "destination", propertyValue = Queues.CHAT_HANDLER),})
@ResourceAdapter("hornetq-ra.rar")
public class ChatHandlerQueueListener implements MessageListener {

 public static final Logger logger = LoggerFactory.getLogger(ChatHandlerQueueListener.class);

 @Inject
 IChatManager chatManager;

 public void onMessage(Message message) {
    ObjectMessage objectMessage = (ObjectMessage) message;
    ComponentMessage routingEngineResponse = null;
    try {
       routingEngineResponse = (ComponentMessage) objectMessage.getObject();
       boolean messageRedelivered = message.getJMSRedelivered();
       if (logger.isTraceEnabled())
          logger.trace("ChatHandlerQueueListener.callingChatManager Incoming response is {}", JsonUtils.toJson(routingEngineResponse));
       if (routingEngineResponse == null)
          return;
       if (messageRedelivered) {
          // Sending the message acknowledgement manually
          message.acknowledge();
       }

    } catch (JMSException e) {
       logger.error("ChatHandlerQueueListener.onMessage Type: null", e);
    }
    if (routingEngineResponse.getType().equals(MessageType.ChatAction) || routingEngineResponse.getType().equals(MessageType.ChatTransfer)) {
       try {
          logger.debug("ChatHandlerQueueListener.callingChatManager {}", JsonUtils.toJson(routingEngineResponse));
          chatManager.processRoutingEngineResponseMessage(routingEngineResponse);
       } catch (UnknownReActionTypeException e) {
          logger.error("ChatHandlerQueueListener.onMessage Type: UnknownReActionTypeException {}", e);
       }
    } else if (routingEngineResponse.getType().equals(MessageType.InboundSms)) {
       logger.debug("Calling request for agent {}", JsonUtils.toJson(routingEngineResponse));
       try {
          chatManager.processChatMessage(routingEngineResponse);
       } catch (ChatServiceUnavailableException | JMSException | ApplicationException e) {
          logger.error("ChatHandlerQueueListener.onMessage Type: Exception {}", e);
       }
    } else if (routingEngineResponse.getType().equals(MessageType.ChatMessage)) {
       try {
          chatManager.processChatMessage(routingEngineResponse);
       } catch (ChatServiceUnavailableException | JMSException | ApplicationException e) {
          logger.error("ChatHandlerQueueListener.onMessage Type: Exception {}", e);
       }
    } else if (routingEngineResponse.getType().equals(MessageType.TropoSmsDelivery)) {
       logger.debug("Calling smsDelvieryHandler {}", JsonUtils.toJson(routingEngineResponse));
       try {
          chatManager.processSmsDeliveryMessage(routingEngineResponse);
       } catch (Exception e) {
          logger.error("ChatHandlerQueueListener.onMessage Type: Exception {}", e);
       }
    } else {
       try {
          logger.trace("Unexpected message seletor found: {}", message.getStringProperty("MESSAGE_TYPE"));
       } catch (JMSException e) {
          logger.error("ChatHandlerQueueListener.onMessage Type: Exception {}", e);
       }
    }
 }

}

JBOSS 配置文件

<hornetq-server>
   <persistence-enabled>true</persistence-enabled>
   <connectors>
      <connector name="netty">
         <factory-class>org.hornetq.core.remoting.impl.netty.NettyConnectorFactory</factory-class>
         <param key="host" value="${jboss.bind.remote.hq.address}"/>
         <param key="port" value="${jboss.bind.remote.hq.port}"/>
      </connector>
   </connectors>
   <jms-connection-factories>
      <connection-factory name="RemoteConnectionFactory">
         <connectors>
            <connector-ref connector-name="netty"/>
         </connectors>
         <entries>
            <entry name="RemoteConnectionFactory"/>
         </entries>
      </connection-factory>
      <pooled-connection-factory name="hornetq-ra">
         <transaction mode="xa"/>
         <connectors>
            <connector-ref connector-name="netty"/>
         </connectors>
         <entries>
            <entry name="java:/JmsXA"/>
         </entries>
      </pooled-connection-factory>
   </jms-connection-factories>
</hornetq-server>

标签： nettyjboss7.xhornetqjboss-mdb

netty - MessageListener 停止监听 HornetQ 队列

问题描述

解决方案

推荐阅读