一、java后台调用MQTT
准备工作：需要导入的jar包
<!-- mqtt依赖包-->
<dependency>
<groupId>org.fusesource.mqtt-client</groupId>
<artifactId>mqtt-client</artifactId>
<version>1.14</version>
</dependency>
1.发布消息
import org.fusesource.hawtbuf.AsciiBuffer;
import org.fusesource.hawtbuf.UTF8Buffer;
import org.fusesource.mqtt.client.Future;
import org.fusesource.mqtt.client.FutureConnection;
import org.fusesource.mqtt.client.MQTT;
import org.fusesource.mqtt.client.QoS;
import java.util.LinkedList;
/**
* MQTT消息发送
* @param topics 主题
* @param data 发送内容
* @throws Exception
*/
public static void checkMQTT(String topics,String data) throws Exception {
String user = env("ACTIVEMQ_USER", "admin"); //mqtt账号
String password = env("ACTIVEMQ_PASSWORD", "admin"); //mqtt密码
String host = env("ACTIVEMQ_HOST", "localhost"); //地址
int port = Integer.parseInt(env("ACTIVEMQ_PORT", "1883")); //端口号
String DATA = data;//发送的信息
String body = "";
for( int i=0; i < DATA.length(); i ++) {
body += DATA.charAt(i%DATA.length());
}
UTF8Buffer bodys = new UTF8Buffer(body); //进行编码转换，防止无法发送中文
MQTT mqtt = new MQTT();
mqtt.setHost(host, port);
mqtt.setUserName(user);
mqtt.setPassword(password);
FutureConnection connection = mqtt.futureConnection();
connection.connect().await();
final LinkedList<Future<Void>> queue = new LinkedList<Future<Void>>();
UTF8Buffer topic = new UTF8Buffer(topics); //对主题进行编码转换，防止中文异常
queue.add(connection.publish(topic, new AsciiBuffer(bodys), QoS.AT_LEAST_ONCE, false)); //发送mqtt
while( !queue.isEmpty() ) {
queue.removeFirst().await();
}
connection.disconnect().await();
}
private static String env(String key, String defaultValue) {
String rc = System.getenv(key);
if( rc== null )
return defaultValue;
return rc;
}
2.订阅消息
public static void getTopic(String topic) throw Exception{
String user = env("ACTIVEMQ_USER", "admin");
String password = env("ACTIVEMQ_PASSWORD", "password");
String host = env("ACTIVEMQ_HOST", "localhost");
int port = Integer.parseInt(env("ACTIVEMQ_PORT", "1883"));
MQTT mqtt = new MQTT();
mqtt.setHost(host, port);
mqtt.setUserName(user);
mqtt.setPassword(password);
final CallbackConnection connection = mqtt.callbackConnection();
connection.listener(new org.fusesource.mqtt.client.Listener() {
public void onConnected() {
}
public void onDisconnected() {
}
public void onFailure(Throwable value) {
value.printStackTrace();
}
public void onPublish(UTF8Buffer utfTopic, Buffer msg, Runnable ack) {
String body = msg.utf8().toString();
if( "SHUTDOWN".equals(body)) {
long diff = System.currentTimeMillis() - start;
System.out.println(String.format("Received %d in %.2f seconds", count, (1.0*diff/1000.0)));
connection.disconnect(new Callback<Void>() {
@Override
public void onSuccess(Void value) {
}
@Override
public void onFailure(Throwable value) {
value.printStackTrace();
}
});
} else {
if( count == 0 ) {
start = System.currentTimeMillis();
}
if( count % 1000 == 0 ) {
System.out.println(String.format("Received %d messages.", count));
}
count ++;
}
ack.run();
}
});
connection.connect(new Callback<Void>() {
@Override
public void onSuccess(Void value) {
Topic[] topics = {new Topic(topic, QoS.AT_LEAST_ONCE)};
connection.subscribe(topics, new Callback<byte[]>() {
public void onSuccess(byte[] qoses) {
}
public void onFailure(Throwable value) {
value.printStackTrace();
}
});
}
@Override
public void onFailure(Throwable value) {
value.printStackTrace();
}
});
}
二、前台页面js调用MQTT
需要引入的js
<script src="js/mqttws31.js"></script>
1.页面接收 和 发送mqtt
//连接mqtt
var client,destination;
function linkMQTT() {
var host = "localhost";
var port = "61614";
var clientId ="example-68312";
var user ="admin";
var password ="admin";
client = new Messaging.Client(host, Number(port), clientId);
client.onConnect = onConnect;
client.onMessageArrived = onMessageArrived;
client.onConnectionLost = onConnectionLost;
client.connect({
userName:user,
cleanSession:true,
password:password,
onSuccess:onConnect,
onFailure:onFailure
});
return false;
}
//接收消息
var onConnect = function(frame) {
client.subscribe('00000001010042'); //订阅主题
client.subscribe('00000001010001');
}
//失败的操作
function onFailure(failure) {
alert(failure.errorMessage);
}
//订阅成功的操作
function onMessageArrived(message) {
alert(message.payloadString);
}
function onConnectionLost(responseObject) {
if (responseObject.errorCode !== 0) {
alert(client.clientId + ": " + responseObject.errorCode + "\n");
}
}
//发送消息
funcation sendMessage(text) {
message = new Messaging.Message(text);
message.destinationName = '00000001010001';
client.send(message);
message = new Messaging.Message(text);
message.destinationName = '00000001010042';
client.send(message);
}