STOMP是一个简单的面向文本的消息传递协议。它定义了一种可互操作的有线格式 ，以便任何可用的 STOMP 客户端都可以与任何 STOMP 消息代理进行通信，从而在语言和平台之间提供简单而广泛的消息互操作性（STOMP 网站有一个STOMP 客户端和服务器实现列表。
代码已上传，查看
使用：
环境准备
RabbitMQ开通STOMP通道：
rebbitmq安装目录sbin文件夹下执行命令
rabbitmq-plugins enable rabbitmq_web_stomp rabbitmq-plugins enable rabbitmq_web_stomp_examples #重启 service rabbitmq-server stop && service rabbitmq-server start
登录web管理页面：http://localhost:15672/ 新建一个交换机
代码
index.html
<!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> <script src="https://code.jquery.com/jquery-3.3.1.min.js"></script> <script src="js\stomp.js"></script> </head> <body> <h1>我的页面</h1> <button> 发送一条消息</button> </body> </html> <script type="module"> import crewebsocket from './js/crewebsocket.js' //连接消息，并监听交换机 crewebsocket.ConnSocket(function(){ $('body').append("<h1>ConnSocket ok !!!</h1>"); crewebsocket.OnOpen(); }); //重写接收消息方法 crewebsocket.onMessageUnit = function(msg){ $('body').append("<h1>" + msg + "</h1>"); } //发送消息 $(document).on('click', 'button',function(){ crewebsocket.SendMsg("hello world"); }); </script>
mywebsocket.js
var mywebsocket = { WsObj: "", ConnUrl: "", msConfig: { base: { rabbitmqServer: "127.0.0.1:15674", rabbitmqUser: "guest", rabbitmqPass: "guest", appid: "zs" }, info: { changename: "testmq_2022", routingKey: "111" } }, InitUrl: function(url, port, userId, userName, unitId, unitName, ip, projId, callback) { if (typeof callback == 'function') { callback(); } }, ConnSocket: function(callback) { this.ConnUrl = "ws://" + this.msConfig.base.rabbitmqServer + "/ws" if ("WebSocket" in window) { this.WsObj = new WebSocket(this.ConnUrl); } else if ("MozWebSocket" in window) { this.WsObj = new MozWebSocket(this.ConnUrl); } if (this.WsObj != "") { this.WsObj = Stomp.over(this.WsObj); this.WsObj.heartbeat.outgoing = 10000; // 客户端每10000ms发送一次心跳检测 this.WsObj.heartbeat.incoming = 0; // // client不接收serever端的心跳检测 this.WsObj.connect(this.msConfig.base.rabbitmqUser, this.msConfig.base.rabbitmqPass, callback, this.OnClose, '/'); this.WsObj.debug = false; //如需调试输出，直接注释此行 this.WsObj.onclose = this.OnClose; this.WsObj.onerror = this.OnError; } // if (typeof callback == 'function') { // callback(); //} }, SendInsideMsg: function(msg) { if (this.WsObj != "" && this.WsObj != undefined) { this.WaitForConnection(function() { mywebsocket.WsObj.send(msg); if (typeof callback == 'function') { callback(); } }, 1000); } }, SendMsg: function(msg, callback) { if (this.WsObj != "" && this.WsObj != undefined) { this.WaitForConnection(function() { mywebsocket.WsObj.send("/exchange/" + mywebsocket.msConfig.info.changename + "/222", {}, msg); if (typeof callback == 'function') { callback(); } }, 1000); } }, WaitForConnection: function(callback, interval) { if (this.WsObj.ws.readyState === 1) { callback(); } else { var that = this; setTimeout(function() { that.WaitForConnection(callback, interval); }, interval); } }, onMessageUnit: function(msg) { //接收新消息 }, onMessagePerson: function(msg) { //接收新消息 }, OnOpen: function() { //连接成功 默认订阅 var exchange1 = "/exchange/" + mywebsocket.msConfig.info.changename + "/" + mywebsocket.msConfig.info.routingKey; this.WsObj.subscribe(exchange1, function(data) { var arr = data.headers.destination.split('/'); if (arr[2] == mywebsocket.msConfig.info.changename) { var msg = data.body; console.log('接收到消息内容: ' + msg); mywebsocket.onMessageUnit(msg); } }); }, OnClose: function(e) { //连接关闭 // this.WsObj = ""; console.log("连接关闭") console.log(e); console.log("开始重连"); mywebsocket.ConnSocket(); }, OnError: function() { //连接错误 //console.log("连接错误") }, CheckBrowser: function() { if (window.WebSocket) { return true == true; } else { return false; } }, BusMsgTri: function(sender, content) { console.log("sender=" + sender + ",content=" + content); } }; //对外暴露的接口 export default mywebsocket;
stomp.js
// Generated by CoffeeScript 1.7.1 /* Stomp Over WebSocket http://www.jmesnil.net/stomp-websocket/doc/ | Apache License V2.0 Copyright (C) 2010-2013 [Jeff Mesnil](http://jmesnil.net/) Copyright (C) 2012 [FuseSource, Inc.](http://fusesource.com) */ (function() { var Byte, Client, Frame, Stomp, __hasProp = {}.hasOwnProperty, __slice = [].slice; Byte = { LF: '\x0A', NULL: '\x00' }; Frame = (function() { var unmarshallSingle; function Frame(command, headers, body) { this.command = command; this.headers = headers != null ? headers : {}; this.body = body != null ? body : ''; } Frame.prototype.toString = function() { var lines, name, skipContentLength, value, _ref; lines = [this.command]; skipContentLength = this.headers['content-length'] === false ? true : false; if (skipContentLength) { delete this.headers['content-length']; } _ref = this.headers; for (name in _ref) { if (!__hasProp.call(_ref, name)) continue; value = _ref[name]; lines.push("" + name + ":" + value); } if (this.body && !skipContentLength) { lines.push("content-length:" + (Frame.sizeOfUTF8(this.body))); } lines.push(Byte.LF + this.body); return lines.join(Byte.LF); }; Frame.sizeOfUTF8 = function(s) { if (s) { return encodeURI(s).match(/%..|./g).length; } else { return 0; } }; unmarshallSingle = function(data) { var body, chr, command, divider, headerLines, headers, i, idx, len, line, start, trim, _i, _j, _len, _ref, _ref1; divider = data.search(RegExp("" + Byte.LF + Byte.LF)); headerLines = data.substring(0, divider).split(Byte.LF); command = headerLines.shift(); headers = {}; trim = function(str) { return str.replace(/^\s+|\s+$/g, ''); }; _ref = headerLines.reverse(); for (_i = 0, _len = _ref.length; _i < _len; _i++) { line = _ref[_i]; idx = line.indexOf(':'); headers[trim(line.substring(0, idx))] = trim(line.substring(idx + 1)); } body = ''; start = divider + 2; if (headers['content-length']) { len = parseInt(headers['content-length']); body = ('' + data).substring(start, start + len); } else { chr = null; for (i = _j = start, _ref1 = data.length; start <= _ref1 ? _j < _ref1 : _j > _ref1; i = start <= _ref1 ? ++_j : --_j) { chr = data.charAt(i); if (chr === Byte.NULL) { break; } body += chr; } } return new Frame(command, headers, body); }; Frame.unmarshall = function(datas) { var frame, frames, last_frame, r; frames = datas.split(RegExp("" + Byte.NULL + Byte.LF + "*")); r = { frames: [], partial: '' }; r.frames = (function() { var _i, _len, _ref, _results; _ref = frames.slice(0, -1); _results = []; for (_i = 0, _len = _ref.length; _i < _len; _i++) { frame = _ref[_i]; _results.push(unmarshallSingle(frame)); } return _results; })(); last_frame = frames.slice(-1)[0]; if (last_frame === Byte.LF || (last_frame.search(RegExp("" + Byte.NULL + Byte.LF + "*$"))) !== -1) { r.frames.push(unmarshallSingle(last_frame)); } else { r.partial = last_frame; } return r; }; Frame.marshall = function(command, headers, body) { var frame; frame = new Frame(command, headers, body); return frame.toString() + Byte.NULL; }; return Frame; })(); Client = (function() { var now; function Client(ws) { this.ws = ws; this.ws.binaryType = "arraybuffer"; this.counter = 0; this.connected = false; this.heartbeat = { outgoing: 10000, incoming: 10000 }; this.maxWebSocketFrameSize = 16 * 1024; this.subscriptions = {}; this.partialData = ''; } Client.prototype.debug = function(message) { var _ref; return typeof window !== "undefined" && window !== null ? (_ref = window.console) != null ? _ref.log(message) : void 0 : void 0; }; now = function() { if (Date.now) { return Date.now(); } else { return new Date().valueOf; } }; Client.prototype._transmit = function(command, headers, body) { var out; out = Frame.marshall(command, headers, body); if (typeof this.debug === "function") { this.debug(">>> " + out); } while (true) { if (out.length > this.maxWebSocketFrameSize) { this.ws.send(out.substring(0, this.maxWebSocketFrameSize)); out = out.substring(this.maxWebSocketFrameSize); if (typeof this.debug === "function") { this.debug("remaining = " + out.length); } } else { return this.ws.send(out); } } }; Client.prototype._setupHeartbeat = function(headers) { var serverIncoming, serverOutgoing, ttl, v, _ref, _ref1; if ((_ref = headers.version) !== Stomp.VERSIONS.V1_1 && _ref !== Stomp.VERSIONS.V1_2) { return; } _ref1 = (function() { var _i, _len, _ref1, _results; _ref1 = headers['heart-beat'].split(","); _results = []; for (_i = 0, _len = _ref1.length; _i < _len; _i++) { v = _ref1[_i]; _results.push(parseInt(v)); } return _results; })(), serverOutgoing = _ref1[0], serverIncoming = _ref1[1]; if (!(this.heartbeat.outgoing === 0 || serverIncoming === 0)) { ttl = Math.max(this.heartbeat.outgoing, serverIncoming); if (typeof this.debug === "function") { this.debug("send PING every " + ttl + "ms"); } this.pinger = Stomp.setInterval(ttl, (function(_this) { return function() { _this.ws.send(Byte.LF); return typeof _this.debug === "function" ? _this.debug(">>> PING") : void 0; }; })(this)); } if (!(this.heartbeat.incoming === 0 || serverOutgoing === 0)) { ttl = Math.max(this.heartbeat.incoming, serverOutgoing); if (typeof this.debug === "function") { this.debug("check PONG every " + ttl + "ms"); } return this.ponger = Stomp.setInterval(ttl, (function(_this) { return function() { var delta; delta = now() - _this.serverActivity; if (delta > ttl * 2) { if (typeof _this.debug === "function") { _this.debug("did not receive server activity for the last " + delta + "ms"); } return _this.ws.close(); } }; })(this)); } }; Client.prototype._parseConnect = function() { var args, connectCallback, errorCallback, headers; args = 1 <= arguments.length ? __slice.call(arguments, 0) : []; headers = {}; switch (args.length) { case 2: headers = args[0], connectCallback = args[1]; break; case 3: if (args[1] instanceof Function) { headers = args[0], connectCallback = args[1], errorCallback = args[2]; } else { headers.login = args[0], headers.passcode = args[1], connectCallback = args[2]; } break; case 4: headers.login = args[0], headers.passcode = args[1], connectCallback = args[2], errorCallback = args[3]; break; default: headers.login = args[0], headers.passcode = args[1], connectCallback = args[2], errorCallback = args[3], headers.host = args[4]; } return [headers, connectCallback, errorCallback]; }; Client.prototype.connect = function() { var args, errorCallback, headers, out; args = 1 <= arguments.length ? __slice.call(arguments, 0) : []; out = this._parseConnect.apply(this, args); headers = out[0], this.connectCallback = out[1], errorCallback = out[2]; if (typeof this.debug === "function") { this.debug("Opening Web Socket..."); } this.ws.onmessage = (function(_this) { return function(evt) { var arr, c, client, data, frame, messageID, onreceive, subscription, unmarshalledData, _i, _len, _ref, _results; data = typeof ArrayBuffer !== 'undefined' && evt.data instanceof ArrayBuffer ? (arr = new Uint8Array(evt.data), typeof _this.debug === "function" ? _this.debug("--- got data length: " + arr.length) : void 0, ((function() { var _i, _len, _results; _results = []; for (_i = 0, _len = arr.length; _i < _len; _i++) { c = arr[_i]; _results.push(String.fromCharCode(c)); } return _results; })()).join('')) : evt.data; _this.serverActivity = now(); if (data === Byte.LF) { if (typeof _this.debug === "function") { _this.debug("<<< PONG"); } return; } if (typeof _this.debug === "function") { _this.debug("<<< " + data); } unmarshalledData = Frame.unmarshall(_this.partialData + data); _this.partialData = unmarshalledData.partial; _ref = unmarshalledData.frames; _results = []; for (_i = 0, _len = _ref.length; _i < _len; _i++) { frame = _ref[_i]; switch (frame.command) { case "CONNECTED": if (typeof _this.debug === "function") { _this.debug("connected to server " + frame.headers.server); } _this.connected = true; _this._setupHeartbeat(frame.headers); _results.push(typeof _this.connectCallback === "function" ? _this.connectCallback(frame) : void 0); break; case "MESSAGE": subscription = frame.headers.subscription; onreceive = _this.subscriptions[subscription] || _this.onreceive; if (onreceive) { client = _this; messageID = frame.headers["message-id"]; frame.ack = function(headers) { if (headers == null) { headers = {}; } return client.ack(messageID, subscription, headers); }; frame.nack = function(headers) { if (headers == null) { headers = {}; } return client.nack(messageID, subscription, headers); }; _results.push(onreceive(frame)); } else { _results.push(typeof _this.debug === "function" ? _this.debug("Unhandled received MESSAGE: " + frame) : void 0); } break; case "RECEIPT": _results.push(typeof _this.onreceipt === "function" ? _this.onreceipt(frame) : void 0); break; case "ERROR": _results.push(typeof errorCallback === "function" ? errorCallback(frame) : void 0); break; default: _results.push(typeof _this.debug === "function" ? _this.debug("Unhandled frame: " + frame) : void 0); } } return _results; }; })(this); this.ws.onclose = (function (_this) { return function () { var msg; msg = "Whoops! Lost connection to " + _this.ws.url; if (typeof _this.debug === "function") { _this.debug(msg); } this.ws = null; _this._cleanUp(); return typeof errorCallback === "function" ? errorCallback(msg) : void 0; }; })(this); return this.ws.onopen = (function(_this) { return function() { if (typeof _this.debug === "function") { _this.debug('Web Socket Opened...'); } headers["accept-version"] = Stomp.VERSIONS.supportedVersions(); headers["heart-beat"] = [_this.heartbeat.outgoing, _this.heartbeat.incoming].join(','); return _this._transmit("CONNECT", headers); }; })(this); }; Client.prototype.disconnect = function(disconnectCallback, headers) { if (headers == null) { headers = {}; } this._transmit("DISCONNECT", headers); this.ws.onclose = null; this.ws.close(); this._cleanUp(); return typeof disconnectCallback === "function" ? disconnectCallback() : void 0; }; Client.prototype._cleanUp = function() { this.connected = false; if (this.pinger) { Stomp.clearInterval(this.pinger); } if (this.ponger) { return Stomp.clearInterval(this.ponger); } }; Client.prototype.send = function(destination, headers, body) { if (headers == null) { headers = {}; } if (body == null) { body = ''; } headers.destination = destination; return this._transmit("SEND", headers, body); }; Client.prototype.subscribe = function (destination, callback, headers) { var client; if (headers == null) { headers = {}; } if (!headers.id) { headers.id = "sub-" + this.counter++; } headers.destination = destination; this.subscriptions[headers.id] = callback; this._transmit("SUBSCRIBE", headers); client = this; return { id: headers.id, unsubscribe: function() { return client.unsubscribe(headers.id); } }; }; Client.prototype.unsubscribe = function(id) { delete this.subscriptions[id]; return this._transmit("UNSUBSCRIBE", { id: id }); }; Client.prototype.begin = function(transaction) { var client, txid; txid = transaction || "tx-" + this.counter++; this._transmit("BEGIN", { transaction: txid }); client = this; return { id: txid, commit: function() { return client.commit(txid); }, abort: function() { return client.abort(txid); } }; }; Client.prototype.commit = function(transaction) { return this._transmit("COMMIT", { transaction: transaction }); }; Client.prototype.abort = function(transaction) { return this._transmit("ABORT", { transaction: transaction }); }; Client.prototype.ack = function(messageID, subscription, headers) { if (headers == null) { headers = {}; } headers["message-id"] = messageID; headers.subscription = subscription; return this._transmit("ACK", headers); }; Client.prototype.nack = function(messageID, subscription, headers) { if (headers == null) { headers = {}; } headers["message-id"] = messageID; headers.subscription = subscription; return this._transmit("NACK", headers); }; return Client; })(); Stomp = { VERSIONS: { V1_0: '1.0', V1_1: '1.1', V1_2: '1.2', supportedVersions: function() { return '1.1,1.0'; } }, client: function(url, protocols) { var klass, ws; if (protocols == null) { protocols = ['v10.stomp', 'v11.stomp']; } klass = Stomp.WebSocketClass || WebSocket; ws = new klass(url, protocols); return new Client(ws); }, over: function(ws) { return new Client(ws); }, Frame: Frame }; if (typeof exports !== "undefined" && exports !== null) { exports.Stomp = Stomp; } if (typeof window !== "undefined" && window !== null) { Stomp.setInterval = function(interval, f) { return window.setInterval(f, interval); }; Stomp.clearInterval = function(id) { return window.clearInterval(id); }; window.Stomp = Stomp; } else if (!exports) { self.Stomp = Stomp; } }).call(this);
测试
启动index页面
去我们的管理页面看看
再看我们的交换机
点击页面上的发送按钮发送几条消息，可以看到已经成功发出去 了
模拟服务端给我们发消息，测试我们的监听是否能正常收到发过来的消息
此时再看我们的页面，已经成功接收并将消息内容追加到页面
