ActiveMQ使用(三):在JavaScript中使用mqttws31.js
ActiveMQ使用(三):在JavaScript中使用mqttws31.js
1. 环境准备
jQuery-1.10
下载地址:https://www.jsdelivr.com/package/npm/jquery-1.10.2?tab=filesmqttws31.js:
下载地址:https://www.jsdelivr.com/package/npm/ng2-mqtt
2. 相关代码
<!DOCTYPE html>
<html lang="en">
<head><meta charset="UTF-8"><meta http-equiv="X-UA-Compatible" content="IE=edge"><meta name="viewport" content="width=device-width, initial-scale=1.0"><title>Document</title>
</head>
<body><div class="connect-input-box"><label for="host">host:</label><input type="text" name="host" placeholder="input host" value="127.0.0.1"><br><label for="port">port:</label><input type="text" name="port" placeholder="input port" value="61614"><br><label for="clientId">client id:</label><input type="text" name="clientId" placeholder="input client id"><br><label for="userId">user id:</label><input type="text" name="userId" placeholder="input user id" value="user"><br><label for="password">password:</label><input type="text" name="password" placeholder="input password" value="pass"><br><label for="destination">destination:</label><input type="text" name="destination" placeholder="input destination" value="world"><br><button id="connect" type="submit">connect</button><button id="disconnect" type="submit">disconnect</button></div><div class="log-box"><p id="log-show"></p></div><div class="send-message-box"><label for="topic">topic:</label><input type="text" name="topic"><br><label for="queue">queue:</label><input type="text" name="queue"><br><input type="text" name="message"><br><button id="send">send</button></div><div class="subscribe-box"><label for="subscribe-topic">subscribe-topic:</label><input type="text" name="subscribe-topic"><button id="subscribe">subscribe</button></div><div class="unsubscribe-box"><label for="unsubscribe-topic"></label><input type="text" name="unsubscribe-topic"><button id="unsubscribe">unsubscribe</button></div><script src="plugins/jquery-1.10.1.js"></script><script src="plugins/mqttws31.js"></script><script>$(() => {$('input[name="clientId"]').val("example-" + Math.floor(Math.random() * 10000))if (!window.WebSocket) {console.log('不支持WebSocket')} else {}})var client, destination$('#connect').click(() => {var host = $('input[name="host"]').val()var port = $('input[name="port"]').val()var clientId = $('input[name="clientId"]').val()var user = $('input[name="userId"]').val()var password = $('input[name="password"]').val()destination = $('input[name="destination"]').val()// 创建一个client 实例client = new Paho.MQTT.Client(host, Number(port), clientId)console.log(client)// 设置回调函数client.onConnectionLost = onConnectionLostclient.onMessageArrived = onMessageArrived// 连接clientclient.connect({onSuccess: onConnect})})// 当client连接时调用function onConnect() {console.log('onConnect')client.subscribe(destination)message = new Paho.MQTT.Message("Hello")message.destinationName = destinationclient.send(message)}// 当客户端断开连接时被调用function onConnectionLost(responseObject) {if (responseObject.errorCode !== 0) {console.log("onConnectionLost: " + responseObject.errorMessage)}}// 当消息发送时调用function onMessageDelivered(message) {console.log('onMessageDelivered: ' + message.payloadString)}// 当消息到达时调用function onMessageArrived(message) {console.log("onMessageArrived:" + message.payloadString);}// 断开连接$('#disconnect').click(() => {console.log('disconnect');client.disconnect()})// 发送消息$('#send').click(() => {console.log('send')let topic = $('input[name="topic"]').val()let payload = $('input[name="message"]').val()let message = new Paho.MQTT.Message(payload)message.destinationName = topicclient.send(message)// client.send("topic", "payload", 2, false)})// 订阅主题$('#subscribe').click(() => {console.log('subscribe')let filter = $('input[name="subscribe-topic"]').val()let subscribeOptions = {qos: 2,invocationContext: {},onSuccess: function(val) { console.log(val) },onFailure: function(val) { console.log(val) },timeout: 60}client.subscribe(filter, subscribeOptions)})// 取消订阅主题$('#unsubscribe').click(() => {console.log('unsubscribe')let filter = $('input[name="unsubscribe-topic"]').val()let subscribeOptions = {invocationContext: {},onSuccess: function(val) { console.log(val) },onFailure: function(val) { console.log(val) },timeout: 60}client.unsubscribe(filter, subscribeOptions)})</script>
</body>
</html>
3. 测试
3.1 连接
3.2 订阅
3.3 发送消息
3.4 取消订阅
3.5 断开连接
4. 相关参考
https://www.eclipse.org/paho/files/jsdoc/index.html
5. 其他
在
SprintBoot项目中集成ActiveMQ后，接收到的数据为字节数组
一种解决方式为:
@JmsListener(destination = "test_producer", containerFactory = "topicListenerContainer")public void receiveTestProducer(Message message) throws JMSException {String msg = StringUtils.activeMQMessageParse(message);System.out.println("收到测试生产者的消息: " + msg);}public class StringUtils {/* 将字符串进行分割并获得字节数组* @param str 待处理字符串* @param split 分割字符串* @return 字节数组*/public static byte[] stringToBytes(String str, String split) {String[] strArr = str.split(split);byte[] byteArr = new byte[strArr.length];for (int i = 0; i < strArr.length; i++) {byteArr[i] = (byte) Integer.parseInt(strArr[i]);}return byteArr;}/* 根据字节数组获取指定字符集的字符串* @param byteArr 字节数组* @param charset 编码字符集* @return 处理后的字符串*/public static String bytesToString(byte[] byteArr, Charset charset) {return new String(byteArr, charset);}/* 将字符串根据分隔符转成字节数组，然后转成指定字符集的字符串* @param str 待处理字符串* @param split 分割字符串* @param charset 指定字符集* @return 处理后的字符串*/public static String stringToString(String str, String split, Charset charset) {return bytesToString(stringToBytes(str, split), charset);}/* 将ActiveMQ接收到的消息转换为UTF-8字符串* @param message* @return*/public static String activeMQMessageParse(Message message) {String str = null;if (message instanceof ActiveMQTextMessage) {ActiveMQTextMessage textMessage = (ActiveMQTextMessage) message;try {str = textMessage.getText().toString();} catch (JMSException e) {e.printStackTrace();}
// System.out.println("text : " + textMessage.getText());} else if (message instanceof ActiveMQBytesMessage) {ActiveMQBytesMessage bytesMessage = (ActiveMQBytesMessage) message;byte[] byteArr = new byte[0];try {byteArr = new byte[(int) bytesMessage.getBodyLength()];int flag = bytesMessage.readBytes(byteArr);str = bytesToString(byteArr, StandardCharsets.UTF_8);
// System.out.println("bytes : " + flag + " : " + str);} catch (JMSException e) {e.printStackTrace();}}return str;}
}
