作为听众的无限循环的替代方案

Question

好吧，也许这是我的一个菜鸟问题。但我想问一下这个。我有一个Java Web应用程序，它通过无限循环从AS400检查DataQueue。如果队列中有消息，它会将其传输到 MQ，如果没有，则继续检查。

起初这是一个好主意，但似乎当我在WAS中部署这个Web应用程序（ServletContextListener）并启动它时，我就是无法阻止它。也许是因为它消耗了资源。

所以也许无限循环不是答案。您知道一种不断检查AS400数据队列上新消息的方法吗？

Answer 1

您不需要经常检查或手动暂停。

您可以将超时值传递给 read（），您的应用程序/线程将在返回之前等待很长时间的条目。 如果你通过-1，它将永远等待...

从
http://pic.dhe.ibm.com/infocenter/iseries/v7r1m0/topic/rzahh/dqconsumerexample.htm#dqconsumerexample

         // Read the first entry off the queue.  The timeout value is
         // set to -1 so this program will wait forever for an entry.
         System.out.println("*** Waiting for an entry for process ***");
         DataQueueEntry DQData = dq.read(-1);
         while (Continue)
         {
            // We just read an entry off the queue.  Put the data into
            // a record object so the program can access the fields of
            // the data by name.  The Record object will also convert
            // the data from server format to Java format.
            Record data = dataFormat.getNewRecord(DQData.getData());
            // Get two values out of the record and display them.
            Integer amountOrdered = (Integer) data.getField("QUANTITY");
            String  partOrdered   = (String)  data.getField("PART_NAME");
            System.out.println("Need " + amountOrdered + " of "
                               + partOrdered);
            System.out.println(" ");
            System.out.println("*** Waiting for an entry for process ***");
            // Wait for the next entry.
            DQData = dq.read(-1);
         }

Answer 2

侦听器可能正在使用所有 CPU，因为它正在尽可能快地循环。如果你限制它 线程.睡眠 ，即使是几毫秒，也会有所帮助。

无论如何，在Java EE环境中，最好使用消息驱动的Bean

http://docs.oracle.com/javaee/6/tutorial/doc/bncgl.html#bncgq

Answer 3

DataQueue read() 方法支持 InterruptedException。

下面是一个使用中断阻塞来处理干净终止的消息循环的示例：

public class DataQueueMappingListener 
  implements ServletContextListener, Runnable {
    public static final int JOIN_TIMEOUT = 30000; // join timeout in millis
    private Thread mapper;
    public void contextInitialized(ServletContextEvent sce) {
        mapper = new Thread(this);
        mapper.setPriority(Thread.MIN_PRIORITY);
        mapper.start();
    }
    public void contextDestroyed(ServletContextEvent sce) {
        mapper.interrupt();
        try {
            mapper.join(JOIN_TIMEOUT);
        } catch (InterruptException e) {
            Thread.currentThread().interrupt();
        }
    }
    public void run() {
        AS400 as400 = new AS400(...);
        DataQueue dq = new DataQueue(as400, ...);
        while(!Thread.currentThread().isInterrupted()) {
            try {
                DataQueueEntry dqe = dq.read();
                if (dqe != null) map(dqe);
            } catch (Exception e) {
                Thread.currentThread().interrupt();
            }
        }
        as400.disconnectService(AS400.DATAQUEUE);
    }
    private void map(DataQueueEntry dqe) {
        // map to MQ here
    }
}

完整示例可作为 GitHub Gist 提供。