SpringBoot與RSocket整合，實(shí)現(xiàn)在線聊天系統(tǒng)
作者：Java知識(shí)日歷
Socket 提供了多種通信模式（Request-Response、Fire-and-Forget、Request-Stream 和 Channel），非常適合實(shí)時(shí)通信場(chǎng)景。相比之下，HTTP/REST 通常用于請(qǐng)求-響應(yīng)模式，不適合長(zhǎng)時(shí)間的連接或頻繁的數(shù)據(jù)交換，可能導(dǎo)致較高的延遲和資源浪費(fèi)。
RSocket 是一個(gè)高性能、雙向通信的二進(jìn)制協(xié)議，適用于實(shí)時(shí)數(shù)據(jù)流和低延遲應(yīng)用場(chǎng)景。
我們?yōu)槭裁催x擇RSocket？
- RSocket 提供了多種通信模式（Request-Response、Fire-and-Forget、Request-Stream 和 Channel），非常適合實(shí)時(shí)通信場(chǎng)景。相比之下，HTTP/REST 通常用于請(qǐng)求-響應(yīng)模式，不適合長(zhǎng)時(shí)間的連接或頻繁的數(shù)據(jù)交換，可能導(dǎo)致較高的延遲和資源浪費(fèi)。
- RSocket 內(nèi)置對(duì)流的支持，可以高效地處理大量并發(fā)連接和數(shù)據(jù)流，適合高并發(fā)的聊天系統(tǒng)。而傳統(tǒng)的 HTTP/REST 需要為每個(gè)請(qǐng)求創(chuàng)建新的線程或連接，高并發(fā)情況下會(huì)導(dǎo)致資源耗盡和性能下降。
- RSocket 是一個(gè)二進(jìn)制協(xié)議，提供了更高的效率和靈活性。而HTTP/REST 使用文本格式（通常是 JSON 或 XML），增加了額外的開銷，并且每個(gè)請(qǐng)求都需要單獨(dú)的連接。
- RSocket 的各個(gè)組件高度模塊化，可以根據(jù)需要進(jìn)行替換和優(yōu)化。
- RSocket 提供了內(nèi)置的安全特性和可靠的消息傳遞機(jī)制。
- RSocket 使用長(zhǎng)連接，減少了連接建立和銷毀的開銷。
- 相比于 HTTP/REST，RSocket 的協(xié)議更加輕量級(jí)，減少了不必要的頭部信息。
哪些公司在使用RSocket？
- Netflix 是 RSocket 的主要貢獻(xiàn)者之一。他們使用 RSocket 來實(shí)現(xiàn)微服務(wù)間的高效通信，特別是在需要實(shí)時(shí)數(shù)據(jù)流和低延遲的應(yīng)用程序中。
- CERN (歐洲核子研究組織)使用 RSocket 來實(shí)現(xiàn)實(shí)時(shí)數(shù)據(jù)分析和監(jiān)控系統(tǒng)。
- Capital One 在其金融應(yīng)用程序中使用 RSocket 來實(shí)現(xiàn)實(shí)時(shí)交易處理和通知系統(tǒng)。
- PayPal 使用 RSocket 來實(shí)現(xiàn)實(shí)時(shí)支付處理和通知系統(tǒng)。
- Uber 使用 RSocket 來實(shí)現(xiàn)實(shí)時(shí)位置跟蹤和調(diào)度系統(tǒng)。
- Intel 使用 RSocket 來實(shí)現(xiàn)實(shí)時(shí)數(shù)據(jù)分析和機(jī)器學(xué)習(xí)模型部署。
- Samsung SDS 在其云服務(wù)和物聯(lián)網(wǎng)解決方案中使用 RSocket。
代碼實(shí)操
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>io.rsocket</groupId>
<artifactId>rsocket-core</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
</dependencies>實(shí)現(xiàn)RSocket服務(wù)端用于處理傳入的消息
package com.example.chat;
import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.FluxSink;
import java.time.Duration;
import java.util.concurrent.ConcurrentHashMap;
@Slf4j
@Controller
publicclass ChatController {
// 存儲(chǔ)所有連接客戶端的FluxSink實(shí)例，用于廣播消息
privatefinal ConcurrentHashMap<String, FluxSink<Message>> clients = new ConcurrentHashMap<>();
/**
* 處理發(fā)送消息請(qǐng)求
*
* @param message 要發(fā)送的消息對(duì)象
* @return 返回空Mono表示操作完成
*/
@MessageMapping("chat.sendMessage")
public Mono<Void> sendMessage(Message message) {
log.info("Received message: {}", message); // 記錄接收到的消息
// 將消息廣播給所有已連接的客戶端
clients.values().forEach(sink -> sink.next(message));
return Mono.empty(); // 操作完成
}
/**
* 處理客戶端連接請(qǐng)求
*
* @param username 客戶端用戶名
* @return 返回一個(gè)Flux流，包含來自服務(wù)器和其他客戶端的消息
*/
@MessageMapping("chat.connect")
public Flux<Message> connect(String username) {
log.info("User connected: {}", username); // 記錄用戶連接事件
// 創(chuàng)建一個(gè)新的Flux流，并將其存儲(chǔ)在clients集合中
return Flux.create(sink -> clients.put(username, sink))
.doOnCancel(() -> { // 當(dāng)客戶端斷開連接時(shí)執(zhí)行的操作
log.info("User disconnected: {}", username); // 記錄用戶斷開連接事件
clients.remove(username); // 從clients集合中移除該用戶的sink
})
.mergeWith(Flux.interval(Duration.ofSeconds(1)) // 合并一個(gè)定時(shí)消息流
.map(tick -> new Message("Server", "Ping"))); // 發(fā)送心跳消息
}
}定義消息類用于傳輸數(shù)據(jù)
package com.example.chat;
import lombok.Data;
/**
* 消息類，用于在客戶端和服務(wù)端之間傳輸消息
*/
@Data
publicclass Message {
private String sender; // 發(fā)送者名稱
private String content; // 消息內(nèi)容
public Message() {}
public Message(String sender, String content) {
this.sender = sender;
this.content = content;
}
}配置RSocket服務(wù)器，接受來自客戶端的連接
package com.example.chat.config;
import io.rsocket.transport.netty.server.TcpServerTransport;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
@Configuration
publicclass RSocketConfig {
/**
* 配置RSocket消息處理器
*
* @param strategies RSocket策略
* @return RSocket消息處理器實(shí)例
*/
@Bean
public RSocketMessageHandler rsocketMessageHandler(RSocketStrategies strategies) {
RSocketMessageHandler handler = new RSocketMessageHandler();
handler.setRSocketStrategies(strategies); // 設(shè)置RSocket策略
handler.route("chat.*") // 設(shè)置路由模式
.acceptMimeType(org.springframework.util.MimeTypeUtils.APPLICATION_JSON); // 設(shè)置支持的消息類型
return handler;
}
/**
* 配置TCP服務(wù)器傳輸方式
*
* @return TCP服務(wù)器傳輸實(shí)例
*/
@Bean
public TcpServerTransport tcpServerTransport() {
return TcpServerTransport.create(7000); // 監(jiān)聽7000端口
}
}測(cè)試
需要先安裝rsc 命令行工具。如果你本地沒有這個(gè)命令工具，請(qǐng)到GitHub (https://github.com/making/rsc/releases) 自行安裝。
第一個(gè)終端窗口A中訂閱消息
rsc --request-stream tcp://localhost:7000 chat.connect -d "Alice"第二個(gè)終端窗口B中發(fā)送消息
rsc --fire-and-forget tcp://localhost:7000 chat.sendMessage -d '{"sender":"Bob","content":"Hello Alice!"}'查看第一個(gè)終端窗口A的結(jié)果
{"sender":"Server","content":"Ping"}
{"sender":"Bob","content":"Hello Alice!"}
{"sender":"Server","content":"Ping"}責(zé)任編輯：武曉燕
來源：
Java知識(shí)日歷
