我正在尝试处理从MQ基础结构获得的一些消息。我有两个阻塞队列sharedQueue和pubQueue。 sharedqueue充满了我从MQ基础设施获得的消息，如下所示，它将消息放入sharedQueue。
client.setCallback（new CallBack（“ inst”，sharedQueue））;
messagemanipulator线程将从sharedQueue读取，对其进行处理，并将响应放入pubQueue，以供以后发布。
新的MessageManipulatorThread（sharedQueue，pubQueue）.run（）;
发布者线程将从pubQueue获取消息，并将其发布到MQ基础结构。
新的PublisherThread（pubQueue）.run（）;
下面是完整的代码：
public class ArrayBlockingQueueExample {
private BlockingQueue<String> sharedQueue = new ArrayBlockingQueue<>(64);
private BlockingQueue<String> pubQueue = new ArrayBlockingQueue<>(64);
public static void main(String[] args) throws MqttException, Exception {
new ArrayBlockingQueueExample().startThreads();
}
public void startThreads() throws MqttException, Exception{
MqttClient client = new MQTTClientFactory().getInstance();
client.setCallback(new CallBack("inst", sharedQueue));
new MessageManipulatorThread(sharedQueue,pubQueue).run();
new PublisherThread(pubQueue).run();
}
public MessageManipulatorThread( BlockingQueue<String> sharedQueue , BlockingQueue<String> pubQueue){
this.sharedQueue = sharedQueue;
this.pubQueue = pubQueue;
}
public void run() {
while (true) {
try {
String msg = sharedQueue.take();
System.out.println(Thread.currentThread().getName() + "manipulator runnning => "+msg);
pubQueue.put(msg);
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
public class PublisherThread implements Runnable {
private BlockingQueue<String> sharedQueue;
public PublisherThread(BlockingQueue<String> sharedQueue){
this.sharedQueue = sharedQueue;
}
public void run() {
while (true) {
System.out.println("Running pub");
try {
System.out.println("pub=>"+sharedQueue.take() );
Thread.sleep(500);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
问题是
new PublisherThread(pubQueue).run();永远无法正常工作。我猜这是一个线程同步问题。pubQueue应该等待直到MessageManipulatorThread填满任何数据，但看起来却不是这样。 >正在等待PublisherThread释放，但是它永远不会变得自由！ ，我还有什么要做的吗？任何帮助深表感谢。 最佳答案
您使用的是Runnable.run()而不是Thread.start()，因此：
new MessageManipulatorThread(sharedQueue,pubQueue).run();
new PublisherThread(pubQueue).run();
不行这是因为
run()实际上在当前线程内运行runnable的方法，而不是创建一个新线程并单独执行它。相反，请执行以下操作：
new Thread(new MessageManipulatorThread(sharedQueue,pubQueue), "MessageManipulatorThread").start();
new Thread(new PublisherThread(pubQueue), "PublisherThread").start();
编辑：
fge在问题中发表了以下评论：
为什么不使用
ExecutorService而不是手工完成它的工作？为了弄清他的意思，他的意思是使用
ExecutorService处理pubQueue的消息，而不是创建一个线程来提取消息并手动处理它们。该代码如下所示：ExecutorService executor = Executors.newSingleThreadExecutor();
new Thread(new MessageManipulatorThread(sharedQueue, executor), "MessageManipulatorThread").start();
然后
MessageManipulatorThread类将更改为：public class MessageManipulatorThread implements Runnable {
private BlockingQueue<String> sharedQueue;
private ExecutorService executor;
public MessageManipulatorThread(BlockingQueue<String> sharedQueue, ExecutorService executor){
this.sharedQueue = sharedQueue;
this.executor = executor;
}
public void run() {
while (true) {
try {
String msg = sharedQueue.take();
System.out.println(Thread.currentThread().getName() + "manipulator runnning => "+msg);
executor.execute(new PublisherThread(msg));
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
然后，您可以更改
PublisherThread，以便它只处理传递给它的单个消息。这是您尝试执行的另一种方法。这种方法还具有一定的灵活性。使用另一种方法，
PublisherThread一次只能处理一个消息（同步）。使用ExecutorService接口允许您更改实现，只需更改以下内容，即可一次（异步）处理多个消息：ExecutorService executor = Executors.newSingleThreadExecutor();
对此：
ExecutorService executor = Executors.newFixedThreadPool(10);
该语句允许执行程序启动最多10个线程，这意味着一次最多可以处理10条消息。有关创建
Executors实现的更多方法，请参见ExecutorService类。