一个生产者,多个消费者
One Producer, multiple Consumers
我一直在编写一些代码,但我需要帮助。
我已经创建了一个生产者和一个消费者,但是我需要创建多个消费者来消费来自生产者的特定 String
,例如我需要一个专门消费 'Move Left Hand'.
的消费者
代码中包含了buffer、producer、consumer和main。我不确定如何通知正确的消费者并比较需要消费的字符串。目前我只有一个消费者。
public class iRobotBuffer {
private boolean empty = true;
public synchronized String take() {
// Wait until message is
// available.
while (empty) {
try {
wait();
} catch (InterruptedException e) {}
}
// Toggle status.
empty = true;
// Notify producer that
// status has changed.
notifyAll();
return message;
}
public synchronized void put(String message) {
// Wait until message has
// been retrieved.
while (!empty) {
try {
wait();
} catch (InterruptedException e) {}
}
// Toggle status.
empty = false;
// Store message.
this.message = message;
// Notify consumer that status
// has changed.
notifyAll();
}
}
public class iRobotConsumer implements Runnable {
private iRobotBuffer robotBuffer;
public iRobotConsumer(iRobotBuffer robotBuffer){
this.robotBuffer = robotBuffer;
}
public void run() {
Random random = new Random();
for (String message = robotBuffer.take();
! message.equals("DONE");
message = robotBuffer.take()) {
System.out.format("MESSAGE RECEIVED: %s%n", message);
try {
Thread.sleep(random.nextInt(5000));
} catch (InterruptedException e) {}
}
}
}
public class iRobotProducer implements Runnable {
private iRobotBuffer robotBuffer;
private int number;
public iRobotProducer(iRobotBuffer robotBuffer)
{
this.robotBuffer = robotBuffer;
//this.number = number;
}
public void run() {
String commandInstructions[] = {
"Move Left Hand",
"Move Right Hand",
"Move Both Hands",
};
int no = commandInstructions.length;
int randomNo;
Random random = new Random();
for (int i = 0;
i < commandInstructions.length;
i++) {
randomNo =(int)(Math.random()*no);
System.out.println(commandInstructions[randomNo]);
robotBuffer.put(commandInstructions[i]);
try {
Thread.sleep(random.nextInt(5000));
} catch (InterruptedException e) {}
}
robotBuffer.put("DONE");
}
}
public class iRobot
{
public static void main(String[] args)
{
iRobotBuffer robotBuffer = new iRobotBuffer();
(new Thread(new iRobotProducer(robotBuffer))).start();
(new Thread(new iRobotConsumer(robotBuffer))).start();
}//main
}//class
问题出在您的 iRobotBuffer class。它需要是一个队列来支持多个生产者/消费者。我已经提供了这样一个队列的代码,但是 java 已经有一个实现 (BlockingDeque<E>
).
public class BlockingQueue<T> {
private final LinkedList<T> innerList = new LinkedList<>();
private boolean isEmpty = true;
public synchronized T take() throws InterruptedException {
while (isEmpty) {
wait();
}
T element = innerList.removeFirst();
isEmpty = innerList.size() == 0;
return element;
}
public synchronized void put(T element) {
isEmpty = false;
innerList.addLast(element);
notify();
}
}
据我了解,您需要 3 个消费者,每个 move 指令一个。
您可以使用 java.util.concurrent
包中的 ArrayBlockingQueue
代替 iRobotBuffer
class。顺便说一下,您可以看看提供的其他并发集合 - 一个可能更适合您。
然后对于消费者,你可以peek()
在队列中的内容上测试它是否符合要求,然后poll()
。
我一直在编写一些代码,但我需要帮助。
我已经创建了一个生产者和一个消费者,但是我需要创建多个消费者来消费来自生产者的特定 String
,例如我需要一个专门消费 'Move Left Hand'.
代码中包含了buffer、producer、consumer和main。我不确定如何通知正确的消费者并比较需要消费的字符串。目前我只有一个消费者。
public class iRobotBuffer {
private boolean empty = true;
public synchronized String take() {
// Wait until message is
// available.
while (empty) {
try {
wait();
} catch (InterruptedException e) {}
}
// Toggle status.
empty = true;
// Notify producer that
// status has changed.
notifyAll();
return message;
}
public synchronized void put(String message) {
// Wait until message has
// been retrieved.
while (!empty) {
try {
wait();
} catch (InterruptedException e) {}
}
// Toggle status.
empty = false;
// Store message.
this.message = message;
// Notify consumer that status
// has changed.
notifyAll();
}
}
public class iRobotConsumer implements Runnable {
private iRobotBuffer robotBuffer;
public iRobotConsumer(iRobotBuffer robotBuffer){
this.robotBuffer = robotBuffer;
}
public void run() {
Random random = new Random();
for (String message = robotBuffer.take();
! message.equals("DONE");
message = robotBuffer.take()) {
System.out.format("MESSAGE RECEIVED: %s%n", message);
try {
Thread.sleep(random.nextInt(5000));
} catch (InterruptedException e) {}
}
}
}
public class iRobotProducer implements Runnable {
private iRobotBuffer robotBuffer;
private int number;
public iRobotProducer(iRobotBuffer robotBuffer)
{
this.robotBuffer = robotBuffer;
//this.number = number;
}
public void run() {
String commandInstructions[] = {
"Move Left Hand",
"Move Right Hand",
"Move Both Hands",
};
int no = commandInstructions.length;
int randomNo;
Random random = new Random();
for (int i = 0;
i < commandInstructions.length;
i++) {
randomNo =(int)(Math.random()*no);
System.out.println(commandInstructions[randomNo]);
robotBuffer.put(commandInstructions[i]);
try {
Thread.sleep(random.nextInt(5000));
} catch (InterruptedException e) {}
}
robotBuffer.put("DONE");
}
}
public class iRobot
{
public static void main(String[] args)
{
iRobotBuffer robotBuffer = new iRobotBuffer();
(new Thread(new iRobotProducer(robotBuffer))).start();
(new Thread(new iRobotConsumer(robotBuffer))).start();
}//main
}//class
问题出在您的 iRobotBuffer class。它需要是一个队列来支持多个生产者/消费者。我已经提供了这样一个队列的代码,但是 java 已经有一个实现 (BlockingDeque<E>
).
public class BlockingQueue<T> {
private final LinkedList<T> innerList = new LinkedList<>();
private boolean isEmpty = true;
public synchronized T take() throws InterruptedException {
while (isEmpty) {
wait();
}
T element = innerList.removeFirst();
isEmpty = innerList.size() == 0;
return element;
}
public synchronized void put(T element) {
isEmpty = false;
innerList.addLast(element);
notify();
}
}
据我了解,您需要 3 个消费者,每个 move 指令一个。
您可以使用 java.util.concurrent
包中的 ArrayBlockingQueue
代替 iRobotBuffer
class。顺便说一下,您可以看看提供的其他并发集合 - 一个可能更适合您。
然后对于消费者,你可以peek()
在队列中的内容上测试它是否符合要求,然后poll()
。