在 MQ v6 api - 如何停止 MQQueue get() 方法？

Question

我正在使用 MQ v6 API 类编写一个简单的 Java 应用程序 现在我可以在 while 循环中浏览远程队列。该程序作为 Windows 服务运行，必须中断然后服务停止。首先我设置 waitInterval=MQC.MQWI_UNLIMITED，但 MQMessage.get() 可以防止循环中断。然后我设置 waitInterval=5000 并捕获 MQRC_NO_MSG_AVAILABLE。这是一个正常的解决方案还是有更好的解决方案？
下面是阅读循环的代码：

public class MessageConsumer {

private MessageListener ml;
private MQQueue queue;
private Thread sideThread;
static Logger logger = Logger.getLogger(MessageConsumer.class);
private static volatile boolean listening = true;

public MessageConsumer(MQQueue queue) {
    this.queue = queue;
}

public void setMessageListener(MessageListener ml) throws MQException {
    this.ml = ml;
    start();
}

public synchronized void stop() {
    this.listening = false;
    logger.log(Priority.INFO, "listening = false");
    sideThread.interrupt(); 
    logger.log(Priority.INFO, "set sideThread.interrupt()");     
}

private void listen() throws MQException {
    MQGetMessageOptions getOptions = new MQGetMessageOptions();
    getOptions.options = MQC.MQGMO_WAIT + MQC.MQGMO_FAIL_IF_QUIESCING + MQC.MQGMO_LOGICAL_ORDER + MQC.MQGMO_ALL_SEGMENTS_AVAILABLE + MQC.MQGMO_COMPLETE_MSG + MQC.MQGMO_SYNCPOINT;
    getOptions.waitInterval = 5000;//MQC.MQWI_UNLIMITED;
    logger.log(Priority.INFO, "Start of listening");
    int i = 1;
    while (listening) {
        //  try {
        System.out.println("Read message");
        MQMessage message = new MQMessage();
        logger.log(Priority.INFO, "Waiting message: ");
        try {
            queue.get(message, getOptions);
             logger.log(Priority.INFO, "Get message: ");
            if (ml != null) {
                ml.onMessage(message);
            }
        } catch (MQException e) {
            if (e.reasonCode == e.MQRC_NO_MSG_AVAILABLE) {
                System.out.println("no more message available or retrived");
            } else {
                throw e;
            }
        }
    }

    logger.log(Priority.INFO, "End of listening");
}

private void start() throws MQException {
    sideThread = new Thread(new Runnable() {

    @Override
    public void run() {
    try {
    listen();
    } catch (MQException mqex) {
    logger.log(Priority.ERROR, "A WebSphere MQ Error occured : Completion Code "
    + mqex.completionCode + " Reason Code "
    + mqex.reasonCode, mqex);

    mqex.printStackTrace();

    }
    }
    });
    sideThread.start();
    try {
    sideThread.join();


    } catch (InterruptedException ex) {
    logger.log(Priority.INFO, "MessageConsumer.start()", ex);
    java.util.logging.Logger.getLogger(MessageConsumer.class.getName()).log(Level.SEVERE, null, ex);
    }
         }

};

Answer 1

使用 5 秒超时不是一个很好的方法 - 我希望你同意这一点，否则你不会问这个问题。

答案是使用线程。在调用get 方法之前，通知线程您正在调用get。当 get 完成时，通知它您已完成。您还必须安排在服务停止时通知线程。当线程被告知服务必须停止时，它应该中断线程（因此您的get）。

这是一个例子。创建其中之一并开始运行。每当您要致电get 时，请致电register()。当get 完成呼叫deRegister()。当服务停止调用stop()，您所有未完成的gets 将被中断。

public class StopWatcher implements Runnable {
  // Use a blocking queue to signal the stop - that way we avoid sleeps etc.
  BlockingQueue stop = new ArrayBlockingQueue(1);
  // All threads that need to be interrupted.
  Set<Thread> needInterrupting = new ConcurrentSkipListSet<Thread> ();

  @Override
  public void run() {
    try {
      // Block on the stop queue - i.e. wait until stop is called.
      stop.take();
    } catch (InterruptedException ex) {
      // Just ignore it - we need to interrupt everyone anyway whether we have been interrupted or not.
    }
    // Interrupt everuone who needs it.
    for ( Thread t : needInterrupting ) {
      t.interrupt();
    }
  }

  // Register for interruption.
  public void register () {
    needInterrupting.add(Thread.currentThread());
  }

  // Register for interruption.
  public void deRegister () {
    needInterrupting.remove(Thread.currentThread());
  }

  // Stop.
  public void stop () {
    // Post something in the queue to trigger the stop process.
    stop.add(this);
  }

}

再想一想 - 如果您已经是多线程的，您可能可以在没有线程的情况下执行此操作。

public class StopWatcher {
  // All threads that need to be interrupted.
  Set<Thread> needInterrupting = new ConcurrentSkipListSet<Thread> ();

  // Register for interruption.
  public void register () {
    needInterrupting.add(Thread.currentThread());
  }

  // Register for interruption.
  public void deRegister () {
    needInterrupting.remove(Thread.currentThread());
  }

  // Stop.
  public void stop () {
    // Interrupt everuone who needs it.
    for ( Thread t : needInterrupting ) {
      t.interrupt();
    }
    needInterrupting.clear();
  }

}

Answer 2

早上好，OldCurmudgeoned。我很困惑。我对您的代码稍作修改：

package threadUtil;

import java.util.Comparator;
import java.util.Set;
import java.util.concurrent.ConcurrentSkipListSet;

public class StopWatcher2 {
    // All threads that need to be interrupted.
//  Set<Thread> needInterrupting = new ConcurrentSkipListSet<Thread> ();

    Set<Thread> needInterrupting = new ConcurrentSkipListSet<Thread>(new Comparator<Thread>() {

        @Override
        public int compare(Thread o1, Thread o2) {
            Long l1 = o1.getId();
            Long l2 = o2.getId();

            if (l1.equals(l2)) {
                return 0;
            } else {
                return -1;
            }
        }
    });

    // Register for interruption.
    public void register() {
        needInterrupting.add(Thread.currentThread());
        System.out.println("register thread: name=" + Thread.currentThread().getId());
    }

    // Register for interruption.
    public void deRegister() {
        needInterrupting.remove(Thread.currentThread());
        System.out.println("deRegister thread: name=" + Thread.currentThread().getId());
    }

    // Stop.
    public void stop() {
        // Interrupt everuone who needs it.
        String name;
        long id;
        for (Thread t : needInterrupting) {
            name = t.getName();
            id = t.getId();
            t.interrupt();
            System.out.println("interrupt thread: name=" + name + ": id= " + id);
        }
        needInterrupting.clear();
    }
}

并编写一个简单的模型（而不是 mqqueue 与套接字一起工作 - 用于建模问题）。所有代码：

package multithreadingportlistener;

import java.io.IOException;
import java.util.Comparator;
import java.util.logging.Level;
import java.util.logging.Logger;
import threadUtil.StopWatcher;


public class MultithreadingPortListener {


    private class ThreadComparator extends Thread implements Comparator<Thread> {

        public ThreadComparator(Thread t) {
        }

        @Override
        public int compare(Thread o1, Thread o2) {
            if (o1.getName().equalsIgnoreCase(o2.getName())) {
                return 0;

            } else {
                return 1;
            }
        }
    }

    public static void main(String[] args) {


        MultiThreadedServer server = new MultiThreadedServer(9000);

        new Thread(server).start();
        try {
            Thread.sleep(20 * 100);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println(
                "Stopping Server");
        server.stop();
    }
}


/*
 * To change this template, choose Tools | Templates
 * and open the template in the editor.
 */
package multithreadingportlistener;

import java.net.ServerSocket;
import java.net.Socket;
import java.io.IOException;
import java.util.logging.Level;
import java.util.logging.Logger;
import threadUtil.StopWatcher2;

public class MultiThreadedServer implements Runnable {

    protected int serverPort = 8080;
    protected ServerSocket serverSocket = null;
    protected boolean isStopped = false;
    protected Thread runningThread = null;
    private StopWatcher2 sw;

    public MultiThreadedServer(int port) {
        this.serverPort = port;
        sw = new StopWatcher2();

    }

    public void run() {
        synchronized (this) {
            this.runningThread = Thread.currentThread();
        }
        openServerSocket();
        while (!isStopped()) {
            Socket clientSocket = null;

            try {
                sw.register();
                clientSocket = this.serverSocket.accept();
                System.out.println("wair accept().");
                sw.deRegister();
            } catch (IOException e) {
                if (isStopped()) {
                    System.out.println("Server Stopped.");
                    return;
                }
                throw new RuntimeException(
                        "Error accepting client connection", e);
            }
            new Thread(
                    new WorkerRunnable(
                    clientSocket, "Multithreaded Server")).start();
        }
        System.out.println("Server Stopped.");
    }

    private synchronized boolean isStopped() {
        return this.isStopped;
    }

    public synchronized void stop() {
        this.isStopped = true;
        sw.stop(); // special instead  this.serverSocket.close() for modelling as mqqueue

    }

    private void openServerSocket() {
        try {
            this.serverSocket = new ServerSocket(this.serverPort);
        } catch (IOException e) {
            throw new RuntimeException("Cannot open port 8080", e);
        }
    }
}

serverSocket.accept() 也作为 MqQueue.get 没有结束。 如何解决这个问题？