ActiveMQ使用(一):在JavaScript中使用stomp.js
ActiveMQ使用(一):在JavaScript中使用stomp.js
1. 环境准备
jQuery-1.10
下载地址:https://www.jsdelivr.com/package/npm/jquery-1.10.2?tab=filesstomp.js 2.3.3:
下载地址:https://www.jsdelivr.com/package/npm/stompjs
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/stomp.js"></script><script type="module">$(() => {console.log('mqtt: ', Stomp)$('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()console.log(host)// 创建一个client 实例let url = 'ws://' + host + ':' + port + '/stomp'client = Stomp.client(url)console.log(client)let headers = {login: user,passcode: password,'client-id': clientId}client.connect(headers, () => {console.log('connect success')client.subscribe(destination, (message) => {if (message.body) {console.log('get body ' + message.body)} else {console.log('got empty message')}})}, () => {console.log('connect failed')})})// 断开连接$('#disconnect').click(() => {console.log('disconnect');client.disconnect(() => {console.log('disconnect')})})// 发送消息$('#send').click(() => {console.log('send')let topic = $('input[name="topic"]').val()let payload = $('input[name="message"]').val()let headers = {}client.send(topic, headers, payload)})var subscribeId = null// 订阅主题$('#subscribe').click(() => {console.log('subscribe')let topic = $('input[name="subscribe-topic"]').val()subscribeId = client.subscribe(topic, (message) => {if (message.body) {console.log('message body: ' + message.body)} else {console.log('empty message')}}, {id: '123123'})})// 取消订阅主题$('#unsubscribe').click(() => {console.log('unsubscribe')console.log(subscribeId);client.unsubscribe(subscribeId.id)})</script>
</body>
</html>
3. 结果展示
3.1 连接
3.2 订阅
3.3 发送消息
注意：默认发送点对点消息, 发送主题消息需要添加前缀/topic
3.4 取消订阅
3.5 断开连接
4. 相关参考
前端 Websocket + Stomp.js 的使用
stompjs的所有方法的使用
activemq BytesMessage || TextMessage
STOMP Over WebSocket
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;}
}