消息未发送以在RTE上回复队列

Question

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

• 批处理工作客户端将消息发送到远程队列
• MDB在此远程队列，过程消息上听和调用EJB
• DLQ在所有重试时都配置为处理通知精疲力尽
• 也应在每个重试时发送通知。避免太多通知，重试间隔足够高

要处理最后一点，我尝试通过在 jmsreplyto 标题中设置 wreate a wreate a 。为了模拟上述流，我创建了以下MDB实现...

主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
   }
}

**独立条目 - full.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(replydestination);      
 LOGGER.info("Reply to:   " + message.getJMSReplyTo());

或发送到重播队列的消息

  replyProducer.send(jmsReplyTo, session.createTextMessage(errorMessage));
  LOGGER.info("exception details sent to reply queue...");