【stomp实战】Stomp协议介绍和客户端的使用

一、STOMP 简介

直接使用WebSocket（或SockJS）就很类似于使用TCP套接字来编写Web应用。因为没有高层级的线路协议（wire protocol），因此就需要我们定义应用之间所发送消息的语义，还需要确保连接的两端都能遵循这些语义。
就像HTTP在TCP套接字之上添加了请求-响应模型层一样，STOMP在WebSocket之上提供了一个基于帧的线路格式（frame-based wire format）层，用来定义消息的语义。
与HTTP请求和响应类似，STOMP帧由命令、一个或多个头信息以及负载所组成。例如，如下就是发送数据的一个STOMP帧：

>>> SEND
transaction:tx-0
destination:/app/marco
content-length:20

{"message":"Marco!"}
1
2
3
4
5
6

在这个例子中，STOMP命令是send，表明会发送一些内容。紧接着是三个头信息：一个表示消息的的事务机制，一个用来表示消息要发送到哪里的目的地，另外一个则包含了负载的大小。然后，紧接着是一个空行，STOMP帧的最后是负载内容。

二、stomp服务端的支持

一些消息中间件对stomp是有支持的。

• RabbitMQ
• HornetQ
• ActiveMQ

以RabbitMQ为例，进行stomp通信示例：参考文档

2.1 rabbitmq开启stomp插件

rabbitmq-plugins enable rabbitmq_web_stomp
1

2.2 rabbitmq使用stomp

开启插件后暴露的端点是

ws://127.0.0.1:15674/ws
1

引入stomp.js，并且使用

<script src="stomp.js"></script>
<script>
var ws = new WebSocket('ws://127.0.0.1:15674/ws');
var client = Stomp.over(ws);
[...]

[...]
var on_connect = function() {
    console.log('connected');
};
var on_error =  function() {
    console.log('error');
};
client.connect('guest', 'guest', on_connect, on_error, '/');
[...]
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

示例代码参考自：https://github.com/rabbitmq/rabbitmq-web-stomp-examples

其中，stomp客户端的支持，查看文档链接，github链接

三、stomp报文的组成

帧格式如下

COMMAND
header1:value1
header2:value2
 
Body^@

//例
SEND   //作为COMMAND
USER-TOKEN:value1     //作为Headers
content-type:application/json   //作为Headers

Hello, stomp.  //消息内容，可以多行，直到^@为止  //作为Body
^@ //此Frame结束
1
2
3
4
5
6
7
8
9
10
11
12
13

示例

>>> SEND
transaction:tx-0
destination:/app/marco
content-length:20

{"message":"Marco!"}
1
2
3
4
5
6

四、stomp客户端使用介绍

stomp客户端的开源包文档在这里，下面简单介绍一下客户端的使用

4.1 创建一个STOMP客户端

4.1.1 通过浏览器中自带的WebSocket来创建

STOMP JavaScript客户端将通过ws:// URL与STOMP服务器进行通信。

要创建一个STOMP客户端JavaScript对象，您需要调用Stomp.client(url)方法，并传入与服务器WebSocket端点相对应的URL：

var url = “ws://localhost:61614/stomp”;
var client = Stomp.client(url);
1
2

上面的示例是stomp.js使用Web浏览器本机WebSocket类来创建WebSocket。

4.1.2 通过自定义的的WebSocket来创建

通过使用Stomp.over(ws)方法，您可以使用其他类型的WebSocket。此方法期望一个符合WebSocket定义的对象。

<script src="http://cdn.sockjs.org/sockjs-0.3.min.js"></script>
  <script>
    // 使用SockJS实现而不是浏览器本机实现
    var ws = new SockJS(url);
    var client = Stomp.over(ws);
    [...]
  </script>
1
2
3
4
5
6
7

在

4.1.3 Node.js应用程序中使用

使用以下命令引入模块：

var Stomp = require(‘stompjs’);
1

要通过TCP套接字连接到STOMP代理，请使用Stomp.overTCP(host, port)方法：

var client = Stomp.overTCP(‘localhost’, 61613);
1

要通过WebSocket连接到STOMP代理，请使用Stomp.overWS(url)方法：

var client = Stomp.overWS(‘ws://localhost:61614/stomp’);
1

除了这种初始化之外，无论是在Web浏览器中还是在其他环境中，STOMP API都保持相同。

4.2 连接服务器

  var connect_callback = function() {
  //定义一个回调方法，当客户端连接上服务器并且认证成功后，会回调
  };

 var error_callback = function(error) {
    // display the error's message header:
    alert(error.headers.message);
  };

  client.connect(login, passcode, connectCallback);
  client.connect(login, passcode, connectCallback, errorCallback);
  client.connect(login, passcode, connectCallback, errorCallback, host);

  client.connect(headers, connectCallback);
  client.connect(headers, connectCallback, errorCallback);


    var headers = {
      login: 'mylogin',
      passcode: 'mypasscode',
      // additional header
      'client-id': 'my-client-id'
    };
    client.connect(headers, connectCallback);

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

断开连接示例如下：

  client.disconnect(function() {
    alert("See you next time!");
  };
1
2
3

4.3 心跳

心跳的作用是连接维护。当stomp服务端支持stomp1.1时，心跳默认是开启的。
心跳配置如下

    client.heartbeat.outgoing = 20000; //每20000ms，客户端发送一次心跳
    client.heartbeat.incoming = 0;     // 客户端不接收服务端的心跳
1
2

4.4 发送消息

//三个参数分别是：发送的目标，类似于kafka中的topic，发送的头，请求体
  client.send("/queue/test", {priority: 9}, "Hello, STOMP");
1
2

如果不想发送任何请求头，仍然需要传参

//请求头为空的情况
 client.send(destination, {}, body);
1
2

4.5 订阅/接收消息

示例如下

var subscription = client.subscribe("/queue/test", callback);

  callback = function(message) {
    // called when the client receives a STOMP message from the server
    if (message.body) {
      alert("got message with body " + message.body)
    } else {
      alert("got empty message");
    }
  });

1
2
3
4
5
6
7
8
9
10
11
12

如果只想接收指定的消息，可以这么写

  var headers = {ack: 'client', 'selector': "location = 'Europe'"};
  client.subscribe("/queue/test", message_callback, headers);
  //仅接收请求头中包含location = 'Europe'的消息
1
2
3

如果想接收多个类型的消息，并且回调方法都是同一个的话，可以这么写

  onmessage = function(message) {
    // called every time the client receives a message
  }
  var sub1 = client.subscribe("queue/test", onmessage);
  var sub2 = client.subscribe("queue/another", onmessage);
1
2
3
4
5

如果不想接收消息了，可以这么写

  var subscription = client.subscribe(...);
  
  ...
  
  subscription.unsubscribe();
1
2
3
4
5

4.6 JSON的支持

注意的是，stomp请求body只支持文本，如果发送/接收内容想传js对象，得先转成json串

  var quote = {symbol: 'APPL', value: 195.46};
  client.send("/topic/stocks", {}, JSON.stringify(quote));

  client.subcribe("/topic/stocks", function(message) {
    var quote = JSON.parse(message.body);
    alert(quote.symbol + " is at " + quote.value);
  };
1
2
3
4
5
6
7

4.7 Acknowledgment

下面的代码是手动Ack情况，如果你了解kafka，会发现这个代码似曾相识。
在订阅时，设置了{ack: ‘client’}，这样就需要我们代码手动提交这个消息的消费结果。

  var subscription = client.subscribe("/queue/test",
    function(message) {
      // do something with the message
      ...
      // and acknowledge it
      message.ack();
    },
    {ack: 'client'}
  );
1
2
3
4
5
6
7
8
9

4.8 事务消息

var tx = client.begin();可以得到一个事务对象，其中有两个方法

• commit() 提交事务
• abort() 回滚事务

在一个事务中发送消息

  // start the transaction
  var tx = client.begin();
  // send the message in a transaction
  client.send("/queue/test", {transaction: tx.id}, "message in a transaction");
  // commit the transaction to effectively send the message
  tx.commit();
1
2
3
4
5
6

4.9 开启调试

通过注册回调方法，输出调试结果

 client.debug = function(str) {
    // append the debug log to a #debug div somewhere in the page using JQuery:
    $("#debug").append(str + "\n");
  };
1
2
3
4

五、总结

• stomp协议是基于websocket的
• stomp的帧格式包含command,header,body三个部分
• 现在支持stomp协议的Server有很多，如rabbitmq,activemq
• stomp协议的客户端有开源的js组件：stomp.js，参考上文
• stomp客户端实现了心跳维护，订阅，取消订阅，发布等功能

六、其它问题

6.1 浏览器websocket与sockJS的区别是什么？

websocket

WebSocket 是 HTML5 开始提供的一种在单个 TCP 连接上进行全双工通讯的协议。WebSocket 使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在 WebSocket API 中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。传统的网站为了实现推送技术，所用的技术都是 Ajax 轮询。

SockJS

有一些浏览器中缺少对WebSocket的支持，而SockJS是一个浏览器的JavaScript库，它提供了一个类似于网络的对象，SockJS提供了一个连贯的，跨浏览器的JavaScriptAPI，它在浏览器和Web服务器之间创建了一个低延迟、全双工、跨域通信通道。SockJS的一大好处在于提供了浏览器兼容性。即优先使用原生WebSocket，如果浏览器不支持WebSocket，会自动降为轮询的方式。

前端使用websocket

<!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>
    <p id="mes"></p>

    <script>
      let pText = document.querySelector("p");
      let ws = new WebSocket("ws://localhost:3000/", ["com.kaazing.echo"]);
      ws.onopen = (e) => {
        console.log("WebSocket open", e);
        setInterval(() => {
          ws.send(new Date().getTime());
        }, 2000);
      };
      ws.onmessage = ({ data }) => {
        pText.innerText = data;
      };
    </script>
  </body>
</html>
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27

前端使用SockJS

值得注意的是，只需要在连接服务器注册端点endPoint时，写访问服务器的全路径URL：

new SockJS('http://127.0.0.1:9091/sbjm-cheng/endpointOyzc');
1