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