我正在使用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);
}
}
};
使用5s超时并不是一个很好的方法——我希望你同意这一点，否则你就不会问这个问题了。
答案是使用线程。在调用get方法之前，通知线程您正在调用get。当get完成时，通知它您已完成。您还必须安排在服务停止时通知线程。当线程被告知服务必须停止时，它应该中断线程(因此你的get)。
这里有一个例子。创建其中一个并启动它。当您要呼叫get时，请呼叫register()。当get完成时，调用deRegister()。当业务停止时，呼叫stop()，您所有未完成的get将被中断。
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();
}
}
早安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();
}
}
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。得到不是结束。如何解决这个麻烦?