netty实现websocket

xiangpei

2024-04-11 ac92608ba4d4a2bbeffc124ee25d5f2778617e0e

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
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
package com.ycl.websocket;
 
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.concurrent.GlobalEventExecutor;
 
import java.util.Objects;
 
 
public class WebSocketHandler extends SimpleChannelInboundHandler {
 
    public static ChannelGroup connects = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
 
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object o) throws Exception {
        if(Objects.nonNull(o) && o instanceof WebSocketFrame){
            this.handleWebSocketFrame(ctx, (WebSocketFrame) o);
        }
    }
 
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("有新的客户端连接上了");
        connects.add(ctx.channel());
    }
 
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("有客户端断开连接了");
        connects.remove(ctx.channel());
    }
 
    // 处理ws数据
    private void handleWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        // 处理关闭连接
        if (frame instanceof CloseWebSocketFrame) {
            connects.remove(ctx.channel());
            ctx.close();
            return;
        }
        if (frame instanceof TextWebSocketFrame) {
            // 处理文本消息
            String text = ((TextWebSocketFrame) frame).text();
            System.out.println("服务器收到客户端数据：" +text);
            // 此处为群发，单独发可使用connects.find(ctx.channel().id()).writeAndFlush()发送
            connects.writeAndFlush(new TextWebSocketFrame("你好客户端"));
            // ...
        } else if (frame instanceof BinaryWebSocketFrame) {
            // 处理二进制消息
            // ...
        } else if (frame instanceof PingWebSocketFrame) {
            // 处理 Ping 消息
            // 收到 Ping 消息，回应一个 Pong 消息（表明我还活着）
            ctx.channel().writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
        } else if (frame instanceof PongWebSocketFrame) {
            // 处理 Pong 消息
            // pong消息如果没有特定需求，不用处理
        } else if (frame instanceof ContinuationWebSocketFrame) {
            // 处理连续帧消息（比较大的数据，分片）
            // ...
        }
    }
 
}