我有一个关于 java 中的多线程的问题。我正在模拟一个链接，其中我有一个生成器、一个缓冲区和一个接收器。所有这些都是使用套接字在它们之间进行通信的独立程序。这工作得很好，但我的缓冲区有点问题。我想通过以下方式模拟一个严格优先级队列，我有两个队列，一个具有高优先级，一个具有低优先级，因此，我有两个缓冲区。我在 BufferContainer 中有这两个，并且 bot 是独立的线程，因为它们每个都从不同的生成器接收数据。问题是我只有一个调度程序，它根据优先级决定将哪些数据包传输到接收器。它似乎正在工作，但我在停止程序时遇到了问题。buffer容器生成两个不同的Buffers和Scheduler。这是三个不同的线程，我需要知道调度程序什么时候完成他的任务。但我不知道该怎么做。这就是我尝试过的：
import java.io.BufferedWriter;
import java.io.FileWriter;
import java.io.IOException;
public class BufferContainer {
private Buffer buffer0;
private Buffer buffer1;
private Scheduler scheduler;
private Queue[] queue;
private Queue q1;
public BufferContainer (int inPort0, int outPort, long queueSize0, double serviceRate0,
int inPort1, long queueSize1, double serviceRate1){
try {
/*Some code here*/
//Scheduler separated thread
this.scheduler = new Scheduler(this.queue, serviceRate0,
serviceRate1, outPort, outLog0, outLog1);
//start the scheduler
this.scheduler.start();
//Buffers separated threads
this.buffer0 = new Buffer(inPort0, queueSize0, this.queue[0], outLog0);
this.buffer1 = new Buffer(inPort1, queueSize1, this.queue[1], outLog1);
//start the buffers
this.buffer0.start();
this.buffer1.start();
/*Some code here*/
while (true){
if(this.scheduler.schedulerTerminated()){
//Some code here
this.scheduler.stop();
break;
}
/*Some code here*/
}
}
//stop Buffers
this.buffer0.stop();
this.buffer1.stop();
/*Some code here*/
} catch (IOException ex) {
}
}
public static void main(String[] args){
BufferContainer buf = new BufferContainer(Integer.parseInt(args[0]), Integer.parseInt(args[1]),
Long.parseLong(args[2]), Double.parseDouble(args[3]), Integer.parseInt(args[4]),
Long.parseLong(args[5]), Double.parseDouble(args[6]));
}
}
我只需要知道调度程序何时结束，并且调度程序内部有一个布尔值，当它发生时设置为 true，我尝试在外部使用 while(true) 来持续检查此变量并在需要时停止调度程序，但是它不工作。关于如何从调度程序通知缓冲区容器的任何想法？
更新
我已经测试了挥发性溶液，但它不起作用。我认为也许我无法在调度程序运行时检查变量。这是我的调度程序的样子：
import java.io.BufferedWriter;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.InetAddress;
import java.net.Socket;
import java.util.Random;
public class Scheduler implements Runnable{
Thread th;
private double serviceRateHigh, serviceRateLow;
private DataOutputStream wr;
private BufferedWriter log0, log1;
private Socket socket;
private String packet;
private Queue highPrioQueue, lowPrioQueue;
//Es possible que es mdifiqui per diferents threads
private volatile boolean shouldStop, schedulerTerminated, buffer0Terminated, buffer1Terminated;
public Scheduler(Queue[] queue, double serviceRateHigh, double serviceRateLow,
int port, BufferedWriter log0, BufferedWriter log1){
//Creem el thread
this.th = new Thread(this);
//Guardem les dues cues
this.setQueues(queue);
this.log0 = log0;
this.log1 = log1;
//Guardem els dos service rate
this.serviceRateHigh = serviceRateHigh;
this.serviceRateLow = serviceRateLow;
this.shouldStop = false;
this.schedulerTerminated = false;
this.buffer0Terminated = false;
this.buffer1Terminated = false;
//Socket y buffer de escriptura
try {
this.socket = new Socket(InetAddress.getByName("127.0.0.1"), port);
this.wr = new DataOutputStream(this.socket.getOutputStream());
} catch (IOException e) {
}
}
//Iniciar el thread
public void start(){
this.th.start();
}
//Parar el thread
public void stop(){
this.shouldStop = true;
}
//When both buffer have notified we can terminate the scheduler
public synchronized boolean schedulerTerminated(){
if (this.buffer0Terminated && this.buffer1Terminated)
this.schedulerTerminated = true;
return this.schedulerTerminated;
}
public synchronized void setBufferTerminated(int n){
if(n == 0) this.buffer0Terminated = true;
if(n == 1) this.buffer1Terminated = true;
}
//Assigna les cues de alta i baixa prioritat
private void setQueues(Queue[] queue){
char prio;
for (int i = 0; i < queue.length; i++){
prio = queue[i].getPrio();
if (prio == 'H')
this.highPrioQueue = queue[i];
else{
if (prio == 'L')
this.lowPrioQueue = queue[i];
}
}
}
//Notifica al thread per a que es desperti
synchronized public void resume(){
notify();
}
protected void sendLastPacketNumberToSink(int pl0, int pl1) {
String datagram = "BYE" + " " + pl0 + " " + pl1;
try {
wr.writeBytes(datagram + '\n');
}catch (IOException e) {}
}
//Proces principal del thread
@Override
public void run(){
while(!shouldStop){
//Creem un generador de nombres aleatoris
Random ran = new Random();
//generem un nombre aleatori entre 0 i 1
double uniform = ran.nextDouble();
if(!this.highPrioQueue.isEmpty()){ //Cua de alta prioritat NO buida
//convertim a exponencial
double sleeptime = Math.log(1-uniform)/(-this.serviceRateHigh);
try{
this.th.sleep(Math.round(sleeptime*1000));
}catch(InterruptedException e){
}
//Enviem el paquet de la cua de alta prioritat
try {
if ((this.packet = this.highPrioQueue.nextPacket()) != null){
this.wr.writeBytes(this.packet + '\n');
log0.write("\nScheduler: Paquet [" + packet + "] enviat desde la cua high\n");
log1.write("\nScheduler: Paquet [" + packet + "] enviat desde la cua high\n");
}
}catch (IOException e) {
}
}
else {
if (!this.lowPrioQueue.isEmpty()){//Cua de alta prioritat buida
//convertim a exponencial
double sleeptime = Math.log(1-uniform)/(-this.serviceRateLow);
try{
this.th.sleep(Math.round(sleeptime*1000));
}catch(InterruptedException e){
}
//Enviem el paquet de la cua de baixa prioritat
try {
if ((this.packet = this.lowPrioQueue.nextPacket()) != null){
this.wr.writeBytes(this.packet + '\n');
log0.write("\nScheduler: Paquet [" + packet + "] enviat desde la cua Low\n");
log1.write("\nScheduler: Paquet [" + packet + "] enviat desde la cua Low\n");
}
}catch (IOException e) {
}
}
}
}
}
}
我不确定是否可以在 run 方法当前正在运行时调用调度程序中的方法。如果发生这种情况，就会出现错误。
谢谢！