Stomp

项目 OkHttps Stomp 基于 OkHttps 的 WebSocket 功能实现了一个非常轻量的 Stomp 客户端协议,它可以让你轻松实现 广播发布与订阅队列发布与订阅 的客户端功能。

添加依赖

要开启 Stomp 功能,需要添加以下依赖:

Maven

<dependency>
     <groupId>cn.zhxu</groupId>
     <artifactId>okhttps-stomp</artifactId>
     <version>4.0.2</version>
</dependency>
1
2
3
4
5

Gradle

implementation 'cn.zhxu:okhttps-stomp:4.0.2'
1

构建 Stomp 实例

Stomp提供静态方法over来构建Stomp实例:

  • over(WHttpTask task) 基于 OkHttps 的 WebSocket 构建一个 Stomp 实例,并自动确认消息
  • over(WHttpTask task, boolean autoAck) 构建一个 Stomp 实例,并可指定是否自动确认消息

例如:

// 使用一个 Websocket 连接构建一个 Stomp 实例,同时设置心跳间隔为 20 秒
Stomp stomp = Stomp.over(OkHttps.webSocket("wss://...").heatbeat(20, 20));
1
2

以上代码构建了一个简单的 Stomp 客户端,并默认在收到消息时会自动确认,如果需要收到确认,可以使用下面的方式:

Stomp stomp = Stomp.over(
        OkHttps.webSocket("wss://...").heatbeat(20, 20),
        false       // 参数设置 autoAck 为 false,将需要手动确认消息
    );
1
2
3
4

连接 Stomp 服务

得到一个Stomp实例后,可使用以下两个方法连接 Stomp 服务器:

  • connect() 直接连接 Stomp 服务器
  • connect(List<Header> headers) 携带一些 Stomp 头信息连接 Stomp 服务器

例如:

stomp.connect();
1

如果服务器在连接时需要指令一些额外信息,比如连接 RabbitMQ 时需要指定loginpasscodevhost:

List<Header> headers = new ArrayList<>();
headers.add(new Header("login", "username"));
headers.add(new Header("passcode", "xxxxxxx"));
headers.add(new Header("host", "your_vhost"));

stomp.connect(headers);
1
2
3
4
5
6

连接状态监听

Stomp.over(OkHttps.webSocket("wss://...").heatbeat(20, 20))
    .setOnConnected(stomp -> {
        // 服务器连接成功回调
    })
    .setOnDisconnected(close -> {
        // 连接已断开回调
    })
    .setOnError(msg -> {
        // 错误监听(v2.4.1 新增)
        // 处理服务器发出的 ERROR 帧
    })
    .setOnException(e -> {
        // 异常监听(v3.1.1 新增)
        // 处理服务器发出的 ERROR 帧
    })
    .connect();
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

另外,你还可以直接使用isConnected()方法类判断连接是否已建立:

// 连接是否已建立
boolean connected = stomp.isConnected();
1
2

消息订阅与退订

订阅广播

stomp.topic("/your-topic", (Message msg) -> {
    // 得到消息负载
    String payload = msg.getPayload();

    // 如果需要手动确认消息,调用此方法确认
    stomp.ack(msg)
});
1
2
3
4
5
6
7

退订广播

stomp.untopic("/your-topic");
1

订阅队列

stomp.queue("/your-queue", (Message msg) -> {
    // 得到消息负载
    String payload = msg.getPayload();

    // 如果需要手动确认消息,调用此方法确认
    stomp.ack(msg)
});
1
2
3
4
5
6
7

退订队列

stomp.unqueue("/your-queue");
1

发送消息

// 发送到广播(since v2.5.0)
stomp.sendToTopic("/your-topic", "Hello World");
// 发送到队列(since v2.5.0)
stomp.sendToQueue("/your-queue", "Hello World");
1
2
3
4

在 v2.5.0 之前,可以使用以下方式`:

// 发送到广播
stomp.sendTo("/topic/your-topic", "Hello World");
// 发送到队列
stomp.sendTo("/queue/your-queue", "Hello World");
1
2
3
4

断开服务

stomp.disconnect();
1

默认的 disconnect 方法会先发送一个 DISCONNECT 消息给服务器,然后等到服务器的确认后再断开连接。这么做的目的是为了确保在真正断开之前,双方已发送的数据尽可能的被对方接收到。

如果由于网络等原因最终导致客户端收不到服务器的确认,那么客户端默认最长等待 10秒,10秒 后无论有无收到确认都会断开连接。当然这个 10 秒也是可以指定的:

// 参数指定客户端等待服务器确认的最大时间,since v3.1.0
stomp.disconnect(int maxWaitSeconds);
1
2

如果确实不需要服务器的确认,想直接断开连接,那么可以使用这个方法:

// 参数指定是否立即断开底层连接,since v3.1.0
stomp.disconnect(boolean immediate);
1
2