未发送到RTE上的应答队列的消息

Question

在使用JBoss EAP6.2的当前应用程序中，有许多由远程EJB调用触发的批处理作业。为了集中处理这些作业的所有通知逻辑，我们决定通过传递序列化消息通过MDB路由所有调用。预期的流程如下：

• 批处理作业客户端向远程队列发送消息。
• MDB侦听此远程队列，处理消息并调用EJB。
• DLQ配置为在所有重试用完时处理通知。
• 还应在每次重试时发送通知。为了避免过多的通知，重试间隔足够高。

为了处理最后一点，我尝试通过在JMSReplyTo头中设置它来创建一个JMSReplyTo队列。为了模拟上面的流程，我创建了下面的MDB实现。

主数据库：

@MessageDriven(name = "MiddleManMDB", activationConfig = {
      @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
      @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/test"),
      @ActivationConfigProperty(propertyName = "connectorClassName", propertyValue = "org.hornetq.core.remoting.impl.netty.NettyConnectorFactory"),
      @ActivationConfigProperty(propertyName = "connectionParameters", propertyValue = "host=localhost;port=5445"),
      @ActivationConfigProperty(propertyName = "user", propertyValue = "queueuser"),
      @ActivationConfigProperty(propertyName = "password", propertyValue = "queuepassword")
})
public class MiddleManMDB implements MessageListener {

   private static final Logger LOGGER = LoggerFactory.getLogger(MiddleManMDB.class);

   @Resource(name = "java:/JmsXA")
   private ConnectionFactory connectionFactory;

   /*
    * (non-Javadoc)
    * @see javax.jms.MessageListener#onMessage(javax.jms.Message)
    */
   @Override
   public void onMessage(Message message)
   {
      try {

         if (message instanceof TextMessage) {
            LOGGER.info("Received text message --> {}", ((TextMessage)message).getText());
         }

         throw new JMSException("thrown exception");
      }
      catch (Exception e) {
         sendToReplyQueue(e.getMessage(), message);

         LOGGER.info("Throwing exception to simulate retry...");
         throw new RuntimeException(e);
      }
   }

   private void sendToReplyQueue(String errorMessage, Message message)
   {
      Context context = null;
      Connection conn = null;

      LOGGER.info("Sending exception details to reply queue...");

      try {
         context = new InitialContext();

         conn = connectionFactory.createConnection();
         Session session = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
         Destination jmsReplyTo = message.getJMSReplyTo();
         MessageProducer replyProducer = session.createProducer(jmsReplyTo);
         replyProducer.send(jmsReplyTo, session.createTextMessage(errorMessage));

      }
      catch (NamingException | JMSException e) {
         e.printStackTrace();
      }
      finally {
         // close connection and context
      }
   }
}

回复MDB:

@MessageDriven(name = "ReplyMDB", activationConfig = {
      @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
      @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/reply")
})
public class ReplyMDB implements MessageListener {

   private static final Logger LOGGER = LoggerFactory.getLogger(ReplyMDB.class);

   @Override
   public void onMessage(Message message) {
      try {
         if (message instanceof TextMessage) {
            LOGGER.info("Received reply message --> " + ((TextMessage)message).getText());
         }
      }
      catch (JMSException e) {
         LOGGER.error("Error in reply queue...", e);
      }
   }
}

**死信MDB:**

@MessageDriven(name = "DeadLetterMDB", activationConfig = {
      @ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue"),
      @ActivationConfigProperty(propertyName = "destination", propertyValue = "queue/dead")
})
public class DeadLetterMDB implements MessageListener {

   private static final Logger LOGGER = LoggerFactory.getLogger(DeadLetterMDB.class);

   @Override
   public void onMessage(Message message) {
      try {
         LOGGER.info("Message has arrived in dead letter queue");
         LOGGER.info("Current delivery count - {}", message.getIntProperty("JMSXDeliveryCount"));

         if (message instanceof TextMessage) {
            LOGGER.info("Received text message --> {}", ((TextMessage)message).getText());
         }
      }
      catch (JMSException e) {
         e.printStackTrace();
      }
   }
}

**客户：**

public static void main(String[] args) {
   Connection connection = null;
   Context context = null;

   try {
      // create context and connection factory

      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
      Destination destination = (Destination) context.lookup("jms/queue/test");
      Destination replyDest = (Destination) context.lookup("jms/queue/reply");
      MessageProducer producer = session.createProducer(destination);
      connection.start();

      TextMessage message = session.createTextMessage("Hello World");
      message.setJMSReplyTo(replyDest);

      producer.send(message);
   }
   catch (NamingException | JMSException e) {
      e.printStackTrace();
   }
   finally {
      // close context and connection
   }
}

**独立的相关条目-Ful.xml：**

<address-settings>
  <address-setting match="jms.queue.testQueue">
      <dead-letter-address>jms.queue.DLQ</dead-letter-address>
      <expiry-address>jms.queue.ExpiryQueue</expiry-address>
      <redelivery-delay>1000</redelivery-delay>
      <max-delivery-attempts>3</max-delivery-attempts>
      <max-size-bytes>10485760</max-size-bytes>
      <address-full-policy>BLOCK</address-full-policy>
      <message-counter-history-day-limit>10</message-counter-history-day-limit>
  </address-setting>
  <address-setting match="jms.queue.replyQueue">
      <redelivery-delay>1000</redelivery-delay>
      <max-delivery-attempts>3</max-delivery-attempts>
      <max-size-bytes>10485760</max-size-bytes>
      <address-full-policy>BLOCK</address-full-policy>
      <message-counter-history-day-limit>10</message-counter-history-day-limit>
  </address-setting>
  <address-setting match="jms.queue.DLQ">
      <redelivery-delay>1000</redelivery-delay>
      <max-delivery-attempts>3</max-delivery-attempts>
      <max-size-bytes>10485760</max-size-bytes>
      <address-full-policy>BLOCK</address-full-policy>
      <message-counter-history-day-limit>10</message-counter-history-day-limit>
  </address-setting>
</address-settings>

<jms-destinations>
  <jms-queue name="testQueue">
      <entry name="queue/test"/>
      <entry name="java:jboss/exported/jms/queue/test"/>
  </jms-queue>
  <jms-queue name="replyQueue">
      <entry name="queue/reply"/>
      <entry name="java:jboss/exported/jms/queue/reply"/>
  </jms-queue>
  <jms-queue name="DLQ">
      <entry name="queue/dead"/>
      <entry name="java:jboss/exported/jms/queue/dead"/>
  </jms-queue>
  <jms-topic name="testTopic">
      <entry name="topic/test"/>
      <entry name="java:jboss/exported/jms/topic/test"/>
  </jms-topic>
</jms-destinations>

现在，在MDB中使用上面的流，在应答队列中永远不会接收消息。所有三个队列都部署在同一台服务器上。

我猜原因是以下几点：

sendToReplyQueue(e.getMessage(), message);
LOGGER.info("Throwing exception to simulate retry...");
throw new RuntimeException(e);

由于发送是异步的，而且我正在抛出RTE (以触发重试)，所以消息从没有发送过。有办法解决这个问题吗？

Answer 1

我猜原因就在下面.

您可以尝试评论RTE。还添加了一些更多的记录器来跟踪。检查“回复目标”设置是否正确。

Message.setJMSReplyTo(回复目的地)；LOGGER.info(“回复：”+ message.getJMSReplyTo())；

或发送到重放队列的消息。

replyProducer.send(jmsReplyTo，session.createTextMessage(errorMessage))；LOGGER.info(“发送给应答队列的异常详细信息.”)；