Spring Boot 集成 SocketIO 实现低延迟双向通信

蹲仔8个月前发布

1500

Socket io 支持客户端和服务器之间的低延迟、双向和基于事件的通信。

引入依赖

在pom.xml中新增相关依赖：

<dependency>
    <groupId>com.corundumstudio.socketio</groupId>
    <artifactId>netty-socketio</artifactId>
    <version>2.0.3</version>
</dependency>

编写配置类和配置文件

@CrossOrigin
@Configuration
public class SocketIOConfig {

    @Value("${socket-server.host}")
    private String SOCKET_HOST;

    @Value("${socket-server.port}")
    private int SOCKET_PORT;
    
    private SocketIOServer server;

    @Bean
    public SocketIOServer socketIOServer() {
        com.corundumstudio.socketio.Configuration config = new com.corundumstudio.socketio.Configuration();
        config.setHostname(SOCKET_HOST);
        config.setPort(SOCKET_PORT);
        server = new SocketIOServer(config);
        server.start();

        return server;
    }

    @PreDestroy
    public void stopSocketIOServer() {
        this.server.stop();
    }

}

其中，@Value("{socket-server.host}") 和 @Value("{socket-server.port}") 属性在 yaml 或者 properties 文件中配置，下面以 yaml 文件为例：

socket-server:
  port: 8085
  host: localhost

该类主要负责配置Socker服务的主机名、端口以及负责启动和停止。

实现连接和断连监听器

当我们启动 Socket 服务器时，将监听我们接下来将创建的事件。

@Service
@Slf4j
public class ChatConnectListener implements ConnectListener {

    @Override
    public void onConnect(SocketIOClient client) {
        log.info("Connected user : {}",client.getSessionId().toString());
    }

}

这里简单实现，当用户连接到服务器的时候，输出客户端的相关信息。

下面再来实现断开链接的监听：

@Service
@Slf4j
public class ChatDisconnectListener implements DisconnectListener {

    @Override
    public void onDisconnect(SocketIOClient client) {
        log.info("Disconnected user : {}",client.getSessionId().toString());
    }

}

这里用来实现用户断开服务器连接时候要处理的逻辑。

实现数据监听器

数据监听器是具体处理业务逻辑的地方。

当数据要被监听时，我们可以先定义传输的数据模型，比如这样：

@Data
public class Message {
    private String username;
    private String content;
    private String targetUsername;
}

定义数据监听器：

@Service
@Slf4j
public class PrivateChatMessageListener implements DataListener<Message> {

    private final SocketIOServer server;

    public PrivateChatMessageListener(SocketIOServer server) {
        this.server = server;
    }

    @Override
    public void onData(SocketIOClient client, Message message, AckRequest ackRequest){
        log.info("User : {}",client.getSessionId().toString());

        // Broadcast the message to all connected clients
        server.getBroadcastOperations().sendEvent("privateChatMessage", message);
    }

}

DataListener 接口的onData方法名将在发送 Message 事件时起作用。该方法会将事件发送到 privateChatMessage 主题，用户正在收听此主题会获取事件。另外，在一个Socket服务器中可以有很多数据监听器。

这里我们编写通过Message实现聊天的业务逻辑：

@Service
@Slf4j
public class ChatService {

    private final SocketIOServer socketIOServer;
    private final ChatConnectListener connectListener;
    private final ChatDisconnectListener disconnectListener;
    private final PrivateChatMessageListener privateChatMessageListener;

    public ChatService(SocketIOServer socketIOServer, ChatConnectListener connectListener, ChatDisconnectListener disconnectListener, PrivateChatMessageListener privateChatMessageListener) {
        this.socketIOServer = socketIOServer;
        this.connectListener = connectListener;
        this.disconnectListener = disconnectListener;
        this.privateChatMessageListener = privateChatMessageListener;
    }

    @PostConstruct
    private void init(){
        socketIOServer.addConnectListener(connectListener);
        socketIOServer.addDisconnectListener(disconnectListener);
        socketIOServer.addEventListener("privateChatMessage", Message.class, privateChatMessageListener);
    }

}

connectListener 被添加到服务器的 connect 监听器
disconnectListener 被添加到服务器的断开连接侦听器
privateChatMessageListener 被添加到服务器的事件监听器中。当客户端向 privateChatMessage 主题发送事件时，sent 事件会触发我们数据监听的 onData 方法。客户端正在侦听 privateChatMessage 主题将接收事件数据。如果有更多的事件侦听器，则应将这些事件侦听器添加到服务器的事件侦听器