package com.chen.mqtt.mosquitto;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
/
* @author: ChenJie
* @date 2018/8/10
*/
public class Publisher {
//tcp://MQTT安装的服务器地址:MQTT定义的端口号
public static final StringHOST =”tcp://172.16.1.86:1883″;
//定义一个主题
public static final StringTOPIC =”speedTopic”;
//定义MQTT的ID，可以在MQTT服务配置中指定
private static final Stringclientid =”server84″;
private MqttClientclient;
private MqttTopictopic;
private StringuserName =”mosquitto”;
private Stringpassword =””;
private MqttMessagemessage;
/
* 构造函数
* @throws MqttException
*/
public Publisher()throws MqttException {
// MemoryPersistence设置clientid的保存形式，默认为以内存保存
client =new MqttClient(HOST, clientid, new MemoryPersistence());
connect();
}
private void connect() {
MqttConnectOptions options =new MqttConnectOptions();
options.setCleanSession(false);
// options.setUserName(userName);
// options.setPassword(password.toCharArray());
//超时时长
options.setConnectionTimeout(100);
//心跳时长
options.setKeepAliveInterval(20);
options.setServerURIs(new String[]{HOST} );
try{
client.setCallback(new PushCallback() );
client.connect(options);
topic =client.getTopic(TOPIC);
}
catch(Exception e){
e.printStackTrace();
}
}
public void publish(MqttTopic topic,MqttMessage message)throws MqttException {
MqttDeliveryToken token = topic.publish(message);
System.out.println(“等待发送成功:”+token.isComplete());
token.waitForCompletion();
System.out.println(“已经发送成功:”+token.isComplete());
}
public static void main(String[] args)throws MqttException {
Publisher server =new Publisher();
server.message =new MqttMessage();
server.message.setQos(1);
server.message.setRetained(true);
for(int i=0;i<10;i++){
server.message.setPayload((“hello,topic speed “+i).getBytes());
server.publish(server.topic,server.message);
}
}
}
发布者：全栈程序员-站长，转载请注明出处：https://javaforall.net/221455.html原文链接：https://javaforall.net
