java - 具有JMS队列的同步使用者

我想以同步方式处理来自Glassfish 3中JMS队列的所有消息，因此我尝试将Glassfish窗口中“ JMS物理目标”中的“最大活动使用者”属性从-1更改为-1。我认为设置此选项后，我将只有一个消费者同时执行OnMessage（）。我遇到的问题是，当我更改该属性时出现此错误：

[I500]: Caught JVM Exception: org.xml.sax.SAXParseException: Content is not allowed in prolog.

[I500]: Caught JVM Exception: com.sun.messaging.jms.JMSException: Content is not allowed in prolog.

sendMessage Error [C4038]: com.sun.messaging.jms.JMSException: Content is not allowed in prolog.

如果有人知道使方法onmessage（）同步的另一种方法，将不胜感激。这是我的消费者类：

@MessageDriven(mappedName = "QueueListener", activationConfig = {
@ActivationConfigProperty(propertyName = "acknowledgeMode", propertyValue = "Auto-acknowledge"),
@ActivationConfigProperty(propertyName = "destinationType", propertyValue = "javax.jms.Queue")
})
public class MessageBean implements MessageListener {



@Override
public void onMessage(Message message) {
    long t1 = System.currentTimeMillis();
    write("MessageBean has received " + message);
    try{

        TextMessage result=(TextMessage)message;
        String text=result.getText();
        write("OTAMessageBean message ID has resolved to " + text);
        int messageID=Integer.valueOf(text);

        AirProcessing aP=new AirProcessing();
        aP.pickup(messageID);


    }
    catch(Exception e){
        raiseError("OTAMessageBean error " + e.getMessage());
    }
   long t2 = System.currentTimeMillis();
   write("MessageBean has finished in " + (t2-t1));

}



}

最佳答案

我遇到了同样的问题，发现的唯一解决方案是设置一个Schedule每十秒钟轮询一次来自队列的消息：

@Stateless
public class MyReceiver {
   @Resource(mappedName = "jms/MyQueueFactory")
   private QueueConnectionFactory connectionFactory;
   @Resource(mappedName = "jms/MyQueue")
   private Queue myQueue;
   private QueueConnection qc;
   private QueueSession session;
   private MessageConsumer consumer;


   @PostConstruct
   void init() {
       try {
         qc = connectionFactory.createQueueConnection();
         session = qc.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);
         consumer = session.createConsumer(myQueue);
         qc.start();
       } catch (JMSException e) {
         throw new RuntimeException(e);
       }
   }

   @PreDestroy
   void cleanup() throws JMSException {
     qc.close();
   }

   @Schedule(hour = "*", minute = "*", second = "*/10", persistent = false)
   public void onMessage() throws JMSException {
     Message message;
     while ((message = consumer.receiveNoWait()) != null) {
       ObjectMessage objMsg = (ObjectMessage) message;
       Serializable content;
       try {
         content = objMsg.getObject();

         //Do sth. with "content" here

         message.acknowledge();
       } catch (JMSException ex) {
         ex.printStackTrace();
       }
    }
  }
}