JavaFX并发和任务（在任务中运行线程）

Question

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

我试图编写一个与一些串行设备（使用JSSC-2.8）接口的小图形用户界面，并根据这些设备的响应更新图形用户界面。但是，在写入消息和设备响应之间有一个延迟，在任意的时间内使用Thread.sleep（）对我来说不是一个可靠的编程方式。因此，我想使用并发包中的等待（）和通知（）方法（包括所有适当的同步），但我不确定如何实现它。我最初所做的是在任务中创建另一个线程，它将编写消息并等待响应，并使用一些绑定来更新图形用户界面。我在最后包含了我的代码。下面是我试图实现的伪代码的一个简短形式：

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 06\r");
                this.wait();
                System.out.println("Power: "+power);
                if(power.equals("OK")){
                    tdk.sendMessage("OUT?\r");
                    this.wait();
                    if(power.equals("OFF")){
                        tdk.sendMessage("OUT ON\r");
                        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.run

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

数据绑定您所描述的内容也可以通过数据绑定实现。有了它，您可以定义一个SimpleStringProperty，它绑定到您的UI标签（.bind（）方法）。如果控制器必须触发其消息，则可以设置StringProperty，UI将自动更新。您描述的场景可以这样使用：

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

我们被告知，级别wait（）/notify（）上的并发通知/等待是非常低的级别。你应该尝试与更高级别的同步方法或帮助者合作（人们已经解决了你的问题：）