共存的 Mqtt-Subscriber 和 UI 崩溃

Question

我或多或少是 Java 编程的新手。我正在开发一个接收 mqtt 消息的程序，该程序几乎每秒都通过经纪人获得收入并具有 UI。我的问题是，如果 mqtt-broker 运行，UI 将无法工作。

其实，我想启动 UI，做一些配置的东西。UI 中有一个页面应该执行一些分析任务，例如捕获通信的当前状态。现在我在 UI 中将 mqtt 代理作为计数器打开 - > UI 阶段消失，只有代理运行。

目标应该是，首先 UI 运行......在下一步中，只捕获一条消息并将其添加到计数器中，UI 在此过程中等待并保持打开（可见）状态。在这个过程之后，broker 休息一下，UI 可以再次运行，并且可能会一次又一次地重复整个过程。

很抱歉这个令人困惑的描述，但也许有人对我的问题有解决方案..谢谢！

Answer 1

这是一个简单的例子public void someLongTask(){    new Thread(()->{        //code for task here        Platform.runLater(()->{            //code to update UI here        });    }).start()}作为注释，我使用 Lambda 表达式来减少代码量

Answer 2

简短的更新...或多或少 Bara'Hashesh 的解决方案工作正常，现在 UI 和 mqtt-broker 共存。现在的问题是新打开的线程是一个“无限”线程，导致 mqtt 消息一次又一次地传入，直到您手动停止它。有没有办法中断新启动的进程。我读过一种方法，mythread.interupt但找不到将它与当前方法结合的方法。

Answer 3

基本上这是 IBM 用于通信的公共脚本。我抛弃了进口和一些评论，它们实际上对问题并不重要。为了回答你的问题马丁，实际上没有包含...公共类 MqttCommunicationClassv1 实现 MqttCallback {/**&nbsp;* The main entry point of the sample.&nbsp;*&nbsp;* This method handles parsing of the arguments specified on the command-line&nbsp;* before performing the specified action.&nbsp;* mqtt_action = publish || subscribe&nbsp;*/public static void mqtt_start(String mqtt_action){&nbsp; &nbsp; // Default settings:&nbsp; &nbsp; boolean quietMode = false;&nbsp; &nbsp; String action =&nbsp; mqtt_action;&nbsp; &nbsp; String topic = "";&nbsp; &nbsp; String message = "Message from client !!!";&nbsp; &nbsp; int qos = 2;&nbsp; &nbsp; String broker = "192.168.100.1";&nbsp; &nbsp; int port = 1883;&nbsp; &nbsp; String clientId = null;&nbsp; &nbsp; String subTopic = "RPI-Measurement";&nbsp; &nbsp; String pubTopic = "RPI-Measurement";&nbsp; &nbsp; boolean cleanSession = true; // Non durable subscriptions&nbsp; &nbsp; boolean ssl = false;&nbsp; &nbsp; String password = null;&nbsp; &nbsp; String userName = null;&nbsp; &nbsp; // Parse the arguments -&nbsp; &nbsp; // Validate the provided arguments&nbsp; &nbsp; if (!action.equals("publish") && !action.equals("subscribe")) {&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("Invalid action: " + action);&nbsp; &nbsp; &nbsp; &nbsp; printHelp();&nbsp; &nbsp; &nbsp; &nbsp; return;&nbsp; &nbsp; }&nbsp; &nbsp; if (qos < 0 || qos > 2) {&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("Invalid QoS: " + qos);&nbsp; &nbsp; &nbsp; &nbsp; printHelp();&nbsp; &nbsp; &nbsp; &nbsp; return;&nbsp; &nbsp; }&nbsp; &nbsp; if (topic.equals("")) {&nbsp; &nbsp; &nbsp; &nbsp; // Set the default topic according to the specified action&nbsp; &nbsp; &nbsp; &nbsp; if (action.equals("publish")) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; topic = pubTopic;&nbsp; &nbsp; &nbsp; &nbsp; } else {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; topic = subTopic;&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; }&nbsp; &nbsp; String protocol = "tcp://";&nbsp; &nbsp; if (ssl) {&nbsp; &nbsp; &nbsp; &nbsp; protocol = "ssl://";&nbsp; &nbsp; }&nbsp; &nbsp; String url = protocol + broker + ":" + port;&nbsp; &nbsp; if (clientId == null || clientId.equals("")) {&nbsp; &nbsp; &nbsp; &nbsp; clientId = "SampleJavaV3_" + action;&nbsp; &nbsp; }&nbsp; &nbsp; // With a valid set of arguments, the real work of&nbsp; &nbsp; // driving the client API can begin&nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; // Create an instance of this class&nbsp; &nbsp; &nbsp; &nbsp; MqttCommunicationClassv1 sampleClient = new MqttCommunicationClassv1(url, clientId, cleanSession, quietMode,&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; userName, password);&nbsp; &nbsp; &nbsp; &nbsp; // Perform the requested action&nbsp; &nbsp; &nbsp; &nbsp; if (action.equals("publish")) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; sampleClient.publish(topic, qos, message.getBytes());&nbsp; &nbsp; &nbsp; &nbsp; } else if (action.equals("subscribe")) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; sampleClient.subscribe(topic, qos);&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; } catch (MqttException me) {&nbsp; &nbsp; &nbsp; &nbsp; // Display full details of any exception that occurs&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("reason " + me.getReasonCode());&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("msg " + me.getMessage());&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("loc " + me.getLocalizedMessage());&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("cause " + me.getCause());&nbsp; &nbsp; &nbsp; &nbsp; System.out.println("excep " + me);&nbsp; &nbsp; &nbsp; &nbsp; me.printStackTrace();&nbsp; &nbsp; }// Private instance variablesprivate MqttClient client;private String brokerUrl;private boolean quietMode;private MqttConnectOptions conOpt;private boolean clean;private String password;private String userName;/**&nbsp;* Constructs an instance of the sample client wrapper&nbsp;*&nbsp;&nbsp;* @param brokerUrl&nbsp; &nbsp; the url of the server to connect to&nbsp;* @param clientId&nbsp; &nbsp; &nbsp;the client id to connect with&nbsp;* @param cleanSession clear state at end of connection or not (durable or&nbsp;*&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp;non-durable subscriptions)&nbsp;* @param quietMode&nbsp; &nbsp; whether debug should be printed to standard out&nbsp;* @param userName&nbsp; &nbsp; &nbsp;the username to connect with&nbsp;* @param password&nbsp; &nbsp; &nbsp;the password for the user&nbsp;* @throws MqttException&nbsp;*/public MqttCommunicationClassv1(String brokerUrl, String clientId, boolean cleanSession, boolean quietMode,&nbsp; &nbsp; &nbsp; &nbsp; String userName, String password) throws MqttException {&nbsp; &nbsp; this.brokerUrl = brokerUrl;&nbsp; &nbsp; this.quietMode = quietMode;&nbsp; &nbsp; this.clean = cleanSession;&nbsp; &nbsp; this.password = password;&nbsp; &nbsp; this.userName = userName;&nbsp; &nbsp; // This sample stores in a temporary directory... where messages temporarily&nbsp; &nbsp; // stored until the message has been delivered to the server.&nbsp; &nbsp; // ..a real application ought to store them somewhere&nbsp; &nbsp; // where they are not likely to get deleted or tampered with&nbsp; &nbsp; String tmpDir = System.getProperty("java.io.tmpdir");&nbsp; &nbsp; MqttDefaultFilePersistence dataStore = new MqttDefaultFilePersistence(tmpDir);&nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; // Construct the connection options object that contains connection parameters&nbsp; &nbsp; &nbsp; &nbsp; // such as cleanSession and LWT&nbsp; &nbsp; &nbsp; &nbsp; conOpt = new MqttConnectOptions();&nbsp; &nbsp; &nbsp; &nbsp; conOpt.setCleanSession(clean);&nbsp; &nbsp; &nbsp; &nbsp; if (password != null) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; conOpt.setPassword(this.password.toCharArray());&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; if (userName != null) {&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; conOpt.setUserName(this.userName);&nbsp; &nbsp; &nbsp; &nbsp; }&nbsp; &nbsp; &nbsp; &nbsp; // Construct an MQTT blocking mode client&nbsp; &nbsp; &nbsp; &nbsp; client = new MqttClient(this.brokerUrl, clientId, dataStore);&nbsp; &nbsp; &nbsp; &nbsp; // Set this wrapper as the callback handler&nbsp; &nbsp; &nbsp; &nbsp; client.setCallback(this);&nbsp; &nbsp; } catch (MqttException e) {&nbsp; &nbsp; &nbsp; &nbsp; e.printStackTrace();&nbsp; &nbsp; &nbsp; &nbsp; log("Unable to set up client: " + e.toString());&nbsp; &nbsp; &nbsp; &nbsp; System.exit(1);&nbsp; &nbsp; }}/**&nbsp;* Publish / send a message to an MQTT server&nbsp;*&nbsp;&nbsp;* @param topicName the name of the topic to publish to&nbsp;* @param qos&nbsp; &nbsp; &nbsp; &nbsp;the quality of service to delivery the message at (0,1,2)&nbsp;* @param payload&nbsp; &nbsp;the set of bytes to send to the MQTT server&nbsp;* @throws MqttException&nbsp;*/public void publish(String topicName, int qos, byte[] payload) throws MqttException {&nbsp; &nbsp; // Connect to the MQTT server&nbsp; &nbsp; log("Connecting to " + brokerUrl + " with client ID " + client.getClientId());&nbsp; &nbsp; client.connect(conOpt);&nbsp; &nbsp; log("Connected");&nbsp; &nbsp; String time = new Timestamp(System.currentTimeMillis()).toString();&nbsp; &nbsp; log("Publishing at: " + time + " to topic \"" + topicName + "\" qos " + qos);&nbsp; &nbsp; // Create and configure a message&nbsp; &nbsp; MqttMessage message = new MqttMessage(payload);&nbsp; &nbsp; message.setQos(qos);&nbsp; &nbsp; // Send the message to the server, control is not returned until&nbsp; &nbsp; // it has been delivered to the server meeting the specified&nbsp; &nbsp; // quality of service.&nbsp; &nbsp; client.publish(topicName, message);&nbsp; &nbsp; // Disconnect the client&nbsp; &nbsp; client.disconnect();&nbsp; &nbsp; log("Disconnected");}/**&nbsp;* Subscribe to a topic on an MQTT server Once subscribed this method waits for&nbsp;* the messages to arrive from the server that match the subscription. It&nbsp;* continues listening for messages until the enter key is pressed.&nbsp;*&nbsp;&nbsp;* @param topicName to subscribe to (can be wild carded)&nbsp;* @param qos&nbsp; &nbsp; &nbsp; &nbsp;the maximum quality of service to receive messages at for&nbsp;*&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; this subscription&nbsp;* @throws MqttException&nbsp;*/public void subscribe(String topicName, int qos) throws MqttException {&nbsp; &nbsp; // Connect to the MQTT server&nbsp; &nbsp; client.connect(conOpt);&nbsp; &nbsp; log("Connected to " + brokerUrl + " with client ID " + client.getClientId());&nbsp; &nbsp; // Subscribe to the requested topic&nbsp; &nbsp; // The QoS specified is the maximum level that messages will be sent to the&nbsp; &nbsp; // client at.&nbsp; &nbsp; // For instance if QoS 1 is specified, any messages originally published at QoS&nbsp; &nbsp; // 2 will&nbsp; &nbsp; // be downgraded to 1 when delivering to the client but messages published at 1&nbsp; &nbsp; // and 0&nbsp; &nbsp; // will be received at the same level they were published at.&nbsp; &nbsp; log("Subscribing to topic \"" + topicName + "\" qos " + qos);&nbsp; &nbsp; client.subscribe(topicName, qos);&nbsp; &nbsp; // Continue waiting for messages until the Enter is pressed&nbsp; &nbsp; log("Press <Enter> to exit");&nbsp; &nbsp; try {&nbsp; &nbsp; &nbsp; &nbsp; System.in.read();&nbsp; &nbsp; } catch (IOException e) {&nbsp; &nbsp; &nbsp; &nbsp; // If we can't read we'll just exit&nbsp; &nbsp; }&nbsp; &nbsp; // Disconnect the client from the server&nbsp; &nbsp; client.disconnect();&nbsp; &nbsp; log("Disconnected");}/**&nbsp;* Utility method to handle logging. If 'quietMode' is set, this method does&nbsp;* nothing&nbsp;*&nbsp;&nbsp;* @param message the message to log&nbsp;*/private void log(String message) {&nbsp; &nbsp; if (!quietMode) {&nbsp; &nbsp; &nbsp; &nbsp; System.out.println(message);&nbsp; &nbsp; }}/****************************************************************//* Methods to implement the MqttCallback interface *//****************************************************************//**&nbsp;* @see MqttCallback#connectionLost(Throwable)&nbsp;*/public void connectionLost(Throwable cause) {&nbsp; &nbsp; // Called when the connection to the server has been lost.&nbsp; &nbsp; // An application may choose to implement reconnection&nbsp; &nbsp; // logic at this point. This sample simply exits.&nbsp; &nbsp; log("Connection to " + brokerUrl + " lost!" + cause);&nbsp; &nbsp; System.exit(1);}/**&nbsp;* @see MqttCallback#deliveryComplete(IMqttDeliveryToken)&nbsp;*/public void deliveryComplete(IMqttDeliveryToken token) {&nbsp; &nbsp; // Called when a message has been delivered to the&nbsp; &nbsp; // server. The token passed in here is the same one&nbsp; &nbsp; // that was passed to or returned from the original call to publish.&nbsp; &nbsp; // This allows applications to perform asynchronous&nbsp; &nbsp; // delivery without blocking until delivery completes.&nbsp; &nbsp; //&nbsp; &nbsp; // This sample demonstrates asynchronous deliver and&nbsp; &nbsp; // uses the token.waitForCompletion() call in the main thread which&nbsp; &nbsp; // blocks until the delivery has completed.&nbsp; &nbsp; // Additionally the deliveryComplete method will be called if&nbsp; &nbsp; // the callback is set on the client&nbsp; &nbsp; //&nbsp; &nbsp; // If the connection to the server breaks before delivery has completed&nbsp; &nbsp; // delivery of a message will complete after the client has re-connected.&nbsp; &nbsp; // The getPendingTokens method will provide tokens for any messages&nbsp; &nbsp; // that are still to be delivered.}/**&nbsp;* @return&nbsp;&nbsp;* @see MqttCallback#messageArrived(String, MqttMessage)&nbsp;*/public void messageArrived(String topic, MqttMessage message) throws MqttException {&nbsp; &nbsp; // Called when a message arrives from the server that matches any&nbsp; &nbsp; // subscription made by the client&nbsp; &nbsp; String time = new Timestamp(System.currentTimeMillis()).toString();&nbsp; &nbsp; System.out.println("Time:\t" + time + "&nbsp; Message:\t" + new String(message.getPayload()));&nbsp; &nbsp; mqttMssg = new String(message.getPayload());&nbsp; &nbsp; System.out.println("Instance 1");}/****************************************************************//* End of MqttCallback methods *//****************************************************************/static void printHelp() {&nbsp; &nbsp; System.out.println("Syntax:\n\n" + "&nbsp; &nbsp; Sample [-h] [-a publish|subscribe] [-t <topic>] [-m <message text>]\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; [-s 0|1|2] -b <hostname|IP address>] [-p <brokerport>] [-i <clientID>]\n\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -h&nbsp; Print this help text and quit\n" + "&nbsp; &nbsp; -q&nbsp; Quiet mode (default is false)\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -a&nbsp; Perform the relevant action (default is publish)\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -t&nbsp; Publish/subscribe to <topic> instead of the default\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; (publish: \"Sample/Java/v3\", subscribe: \"Sample/#\")\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -m&nbsp; Use <message text> instead of the default\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; (\"Message from MQTTv3 Java client\")\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -s&nbsp; Use this QoS instead of the default (2)\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -b&nbsp; Use this name/IP address instead of the default (m2m.eclipse.org)\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -p&nbsp; Use this port instead of the default (1883)\n\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -i&nbsp; Use this client ID instead of SampleJavaV3_<action>\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -c&nbsp; Connect to the server with a clean session (default is false)\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; &nbsp;\n\n Security Options \n" + "&nbsp; &nbsp; &nbsp;-u Username \n" + "&nbsp; &nbsp; &nbsp;-z Password \n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; &nbsp;\n\n SSL Options \n" + "&nbsp; &nbsp; -v&nbsp; SSL enabled; true - (default is false) "&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -k&nbsp; Use this JKS format key store to verify the client\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -w&nbsp; Passpharse to verify certificates in the keys store\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "&nbsp; &nbsp; -r&nbsp; Use this JKS format keystore to verify the server\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + " If javax.net.ssl properties have been set only the -v flag needs to be set\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "Delimit strings containing spaces with \"\"\n\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "Publishers transmit a single message then disconnect from the server.\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "Subscribers remain connected to the server and receive appropriate\n"&nbsp; &nbsp; &nbsp; &nbsp; &nbsp; &nbsp; + "messages until <enter> is pressed.\n\n");}}