JavaFX并发和任务(在Task中运行线程)

Question

我是JavaFx/Concurrency的新手，所以我阅读了JavaFx并发教程，但我仍然对JavaFx Gui中后台线程的实现有点困惑。

我试图写一个小的GUI接口与一些串行设备(使用JSSC-2.8)，并根据这些设备的响应更新GUI。但是，在写入消息和设备响应之间存在延迟，并且在任意时间内使用Thread.sleep()对我来说不是一种可靠的编程方式。因此，我想使用并发包中的wait()和notify()方法(具有所有适当的同步)，但我不确定如何实现它。我最初所做的是在Task内部创建另一个线程，它将编写消息并等待响应，并使用一些绑定来更新GUI。我把我的代码放在了最后。以下是我试图实现的伪代码的简短形式:

start Task:
  connect to serial devices
  synchronized loop: 
    send messages
    wait() for event to fire
      notify()

但是发生的事情是，一旦我调用wait()，整个应用程序空闲，然后当调用notify()时(在响应触发和事件之后)，它不会继续在recipe()循环或startTdk()循环中停止，它只是空闲。我是否错误地实现了线程?当我调用wait()时，是否有可能导致EventDispatch或JavaFX应用程序线程暂停?

我希望问题是清楚的，如果有任何澄清需要我可以更新帖子。

public class OmicronRecipe extends Service<String> implements Runnable{
private final String SEPERATOR=";";
private final Tdk tdk;
private final Pvci pvci;
private final SimpleStringProperty data = new SimpleStringProperty(""); 
private final Float MAX_V = 26.0f,UHV=1e-8f;
private boolean isTdkOn=false, isPvciOn=false;
private String power;
private Float temp,press,maxT, setT;
private int diffMaxT,diffP,diffPow, diffT, index=0;
public OmicronRecipe(){
    tdk = new Tdk("COM4");
    pvci = new Pvci("COM5");
}
private synchronized void recipe(){
        while (true){
            try {
                sendMessages();
                data.set(power+SEPERATOR+temp+SEPERATOR+press);
                calcDiffs();
                if (diffPow < 0){
                    if(diffMaxT < 0){
                        if(diffT < 0){
                            if (diffP < 0){
                                if(!rampPow()){
                                    //Max Power reached
                                }
                            }else{
                                //Wait for pressure drop
                            }
                        }
                    }else{
                        //Wait until quit
                    }
                }else{
                    //Max power reached
                }
                Thread.sleep(5000);
            } catch (InterruptedException ex) {
                Logger.getLogger(OmicronRecipe.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
}
private synchronized boolean rampPow(){
    boolean isRamped=false;
    Float setPow = tdk.getSetPow(index), curPow;
    setT = tdk.getSetT(index);
    curPow = Float.parseFloat(power);
    if(curPow.compareTo(setPow) < 0){
        do{
            curPow += 0.1f;
            tdk.sendMessage("PV "+curPow+"r");
            try {
                wait();
            } catch (InterruptedException ex) {
                Logger.getLogger(OmicronRecipe.class.getName()).log(Level.SEVERE, null, ex);
            }
            curPow = Float.parseFloat(power);
        }while(curPow.compareTo(setPow) < 0);
        index++;
        isRamped=true;
    }
    return isRamped;
}
public synchronized boolean connect(){
    if(!isTdkOn && !isPvciOn){
        isTdkOn = tdk.connect();
        isPvciOn = pvci.connect();
    }
    return isTdkOn && isPvciOn;
}
public synchronized boolean disconnect(){
    if(tdk!=null && pvci !=null){
        isTdkOn = tdk.disconnect();
        isPvciOn = pvci.disconnect();
    }
    return !isTdkOn && !isPvciOn;
}
public synchronized StringProperty getData(){
    return data;
}
public void setMaxT(Float maxT){
    this.maxT = maxT;
}
private synchronized void calcDiffs(){
    Float pow = Float.parseFloat(power);
    diffPow = pow.compareTo(MAX_V);
    diffMaxT = temp.compareTo(maxT);
    diffT = temp.compareTo(100f);
    diffP = press.compareTo(UHV);
}
private synchronized void setListeners(){
    tdk.getLine().addListener((ov,t, t1)-> {
        synchronized (this){
            System.out.println("New Power: "+t1);
            power = t1;
            this.notify();
        }
    });
    pvci.getLine().addListener((ov,t,t1) ->{
        synchronized (this){
        String[] msg = t1.split(SEPERATOR);
        if(msg.length == 2){
            switch(msg[0]){
                case "temperature":
                    System.out.println("Temperaute");
                    temp = Float.parseFloat(msg[1]);
                    break;
                case "pressure":
                    System.out.println("Pressure");
                    press = Float.parseFloat(msg[1]);
                    break;
                default:
                    System.out.println("Nothing; Something went wrong");
                    break;
            }
        }
            this.notify();
        }
    });
}
private synchronized void sendMessages(){
        try {
            tdk.sendMessage("PV?r");
            this.wait();
            pvci.sendMessage("temperature");
            this.wait();
            pvci.sendMessage("pressure");
            this.wait();
        } catch (InterruptedException ex) {
            Logger.getLogger(OmicronRecipe.class.getName()).log(Level.SEVERE, null, ex);
        }
}
private synchronized boolean startTdk(){
    boolean isOut=false;
        if(isTdkOn){
            try {
                tdk.sendMessage("ADR 06r");
                this.wait();
                System.out.println("Power: "+power);
                if(power.equals("OK")){
                    tdk.sendMessage("OUT?r");
                    this.wait();
                    if(power.equals("OFF")){
                        tdk.sendMessage("OUT ONr");
                        this.wait();
                        isOut = power.equals("ON");
                    }
                    else{
                        isOut = power.equals("ON");
                    }
                }
            } catch (InterruptedException ex) {
                Logger.getLogger(OmicronRecipe.class.getName()).log(Level.SEVERE, null, ex);
            }
        }
        return isOut;
}
@Override
protected Task<String> createTask() {
    return new Task<String>() {
          @Override
          protected String call() throws IOException{
            new Thread(new OmicronRecipe()).start();
            return "";
          }
      };
}
@Override
public void run() {
    if (connect()){
        setListeners();
        if(startTdk()){
            recipe();
        }
    }
}
}

我将不包括Pvci类，因为它只是Tdk类的副本，但具有特定的消息序列以与该机器通信。

public class Tdk {
private SerialPort tdkPort;
private final String portName;
private StringBuilder sb = new StringBuilder("");;
private final StringProperty line = new SimpleStringProperty("");
private final HashMap<Float,Float> calibMap;
private ArrayList<Float> list ;
private boolean isEnd=false;
public Tdk(String portName){
    this.portName = portName;
    System.out.println("TDK at "+portName);
    calibMap = new HashMap();
    setMap();
}
public synchronized boolean connect(){
    tdkPort = new SerialPort(portName);
    try {
        System.out.println("Connecting");
        tdkPort.openPort();
        tdkPort.setParams(9600,
                SerialPort.DATABITS_8,
                SerialPort.STOPBITS_1,
                SerialPort.PARITY_NONE);
        tdkPort.setEventsMask(SerialPort.MASK_RXCHAR);
        tdkPort.addEventListener(event -> {
            if(event.isRXCHAR()){
                if(event.getPortName().equals(portName)){
                    try {
                        if(!isEnd){
                            int[] str = tdkPort.readIntArray();
                            if(str!=null)
                                hexToString(str);    
                        }
                        if(isEnd){
                            System.out.println("Here: "+sb.toString());
                            isEnd=false;
                            String d = sb.toString();
                            sb = new StringBuilder("");
                            line.setValue(d);
                        }
                    } catch (SerialPortException e) {
                            Logger.getLogger(Tdk.class.getName()).log(Level.SEVERE, null, e);
                    }
                }
            }
        });
    } catch (SerialPortException e) {
            Logger.getLogger(Tdk.class.getName()).log(Level.SEVERE, null, e);
    }
    return tdkPort !=null && tdkPort.isOpened();
}
public synchronized boolean disconnect(){
    if(tdkPort!=null) {
        try {
            tdkPort.removeEventListener();
            if (tdkPort.isOpened())
                    tdkPort.closePort();
        } catch (SerialPortException e) {
                Logger.getLogger(Tdk.class.getName()).log(Level.SEVERE, null, e);
        }
        System.out.println("Disconnecting");
    }
    return tdkPort.isOpened();
}
public synchronized void sendMessage(String message){
    try {
        tdkPort.writeBytes(message.getBytes());
    } catch (SerialPortException e) {
            Logger.getLogger(Tdk.class.getName()).log(Level.SEVERE, null, e);
    }
}
private void setMap(){
    calibMap.put(1.0f, 25.0f);
    calibMap.put(7.0f, 125.0f);
    calibMap.put(9.8f, 220.0f);
    list = new ArrayList(calibMap.keySet());
}
public Float getSetPow(int index){
    return list.get(index);
}
public Float getSetT(int index){
    return calibMap.get(list.get(index));
}
public synchronized StringProperty getLine(){
    return line;
}
private synchronized void hexToString(int[] hexVal){
    for(int i : hexVal){
        if(i != 13){
            sb.append((char)i);
        }else{
            isEnd=true;
        }
    }
    System.out.println("Turning: "+Arrays.toString(hexVal)+" to String: "+sb.toString()+" End: "+isEnd);
}

Answer 1

冻结

你的UI冻结最有可能是因为你正在等待FX应用程序线程，解决这个问题有不同的方法:

---

JavaFX应用线程

您可以将一些工作委托给FX应用程序线程，因此请参阅Platform.runLater

不是所有的东西都可以在这个线程上运行，但是，例如，在你的DeviceController中，你可以等到消息出现，然后调用Platform.runLater()并更新字段(因此你应该将字段移交给控制器)。

---

绑定你所描述的也可以通过数据绑定来实现。这样你就可以定义一个SimpleStringProperty，它被绑定到你的UI Label (.bind() Method)上。如果控制器必须触发它的消息，你可以设置StringProperty, UI就会自我更新。您所描述的场景可以这样使用:

start Task:
    connect to serial devices
    synchronized loop: 
        send messages
        wait() for event to fire
        **updateDate the DataBounded fields**

---

我们被教导，并发通知/等待级别wait()/notify()的并发性非常低。您应该尝试使用更高级别的同步方法或帮助器(人们已经解决了您的问题:))