上万字全面解读websocket(多种实现方案，含集群实现代码)

Websocket

提供多种websocket实现方案，包括集群模式的解决，附详细代码，轻松掌握websocket。

总结

• Websocket介绍
• Websocket使用场景
• 基于SpringBoot实现Websocket
• 基于Netty实现Websocket
• 基于tio实现Websocket(推荐)
• 实现websocket(集群版)
• demo代码

Websocket介绍

WebSocket是一种在单个TCP连接上进行全双工通信的协议。WebSocket使得客户端和服务器之间的数据交换变得更加简单，允许服务端主动向客户端推送数据。在WebSocket API中，浏览器和服务器只需要完成一次握手，两者之间就直接可以创建持久性的连接，并进行双向数据传输。

Websocket使用场景

• 数字大屏实时更新，典型的如：航班信息，股票基金报价，体育实况等
• 消息提醒
• 社交订阅
• 多人聊天
• web页面日志实时查看
• web页面中模拟shell交互
• 等等

基于SpringBoot实现WS

引入依赖

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-websocketartifactId>
dependency>

<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-webartifactId>
dependency>
1
2
3
4
5
6
7
8
9

添加Webscoket配置

/**
 * 自动注册使用@ServerEndpoint注解声明的websocket endpoint
 */
@Component
public class WebSocketConfig {

    @Bean
    public ServerEndpointExporter serverEndpointExporter() {
        return new ServerEndpointExporter();
    }
}
1
2
3
4
5
6
7
8
9
10
11

Webscoket通信代码

@Component
@ServerEndpoint("/websocket/{id}")
@Slf4j
public class WebSocketServer {


    /**
     *  连接建立成功后调用
     */
    @OnOpen
    public void onOpen(@PathParam(value = "id") String id, Session session) {
        log.info("客户端" + id + "连接建立.");
        WsSessionManager.add(id, session);

        try {
            sendMessage(id, "客户端" + id + "连接建立.");
        } catch (IOException e) {
            log.error("WebSocket IO异常");
        }
    }

    /**
     * 连接关闭时调用
     */
    @OnClose
    public void onClose(@PathParam(value = "id") String id, Session session) {
        log.info("有一连接关闭：{}", id);
        WsSessionManager.remove(id);
    }

    /**
     * 收到客户端消息后调用
     */
    @OnMessage
    public void onMessage(@PathParam(value = "id") String id, String message) {
        log.info("来自客户端的消息:" + message);
        String[] messages = message.split("[|]");
        try {
            if (messages.length > 1) {
                sendToUser(messages[0], messages[1], id);
            } else {
                sendToAll(messages[0]);
            }
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
    }

    /**
     * 发生错误时回调
     */
    @OnError
    public void onError(Session session, Throwable e) {
        log.error("WebSocket发生错误:{}", e.getMessage(), e);
    }

    /**
     * 发送消息
     * @param message 要发送的消息
     */
    private void sendMessage(String id, String message) throws IOException {
        Session session = WsSessionManager.get(id);
        session.getBasicRemote().sendText(message);
    }

    private void sendToUser(String message, String sendClientId, String myId) throws IOException {
        if (sendClientId == null || WsSessionManager.get(sendClientId) == null) {
            sendMessage(myId, "当前客户端不在线");
        } else {
            sendMessage(sendClientId, message);
        }

    }

    private void sendToAll(String message) throws IOException {
        for (String key : WsSessionManager.SESSION_POOL.keySet()) {
            WsSessionManager.get(key).getBasicRemote().sendText(message);
        }
    }
}
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
68
69
70
71
72
73
74
75
76
77
78
79
80

Session管理管理工具类

/**
 * Session管理管理工具类
 */
@Slf4j
public class WsSessionManager {

    /**
     * 保存连接 session 的地方
     */
    public static ConcurrentHashMap<String, Session> SESSION_POOL = new ConcurrentHashMap<>();

    /**
     * 添加 session
     *
     * @param key
     */
    public static void add(String key, Session session) {
        // 添加 session
        SESSION_POOL.put(key, session);
    }

    /**
     * 删除 session,会返回删除的 session
     *
     * @param key
     * @return
     */
    public static Session remove(String key) {
        // 删除 session
        return SESSION_POOL.remove(key);
    }

    /**
     * 删除并同步关闭连接
     *
     * @param key
     */
    public static void removeAndClose(String key) {
        Session session = remove(key);
        if (session != null) {
            try {
                // 关闭连接
                session.close();
            } catch (IOException e) {
                log.error("删除并同步关闭连接异常:{}", e.getMessage(), e);
            }
        }
    }

    /**
     * 获得 session
     *
     * @param key
     * @return
     */
    public static Session get(String key) {
        // 获得 session
        return SESSION_POOL.get(key);
    }
}
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

测试

这里推荐一个在线的测试工具：http://coolaf.com/zh/tool/chattest

你发送的信息 2022-09-04 14:05:41
你好|4
你发送的信息 2022-09-04 14:05:57
你好|4
websocket连接已断开!!!
连接成功，现在你可以发送信息啦！！！
服务端回应 2022-09-04 14:09:23
客户端1连接建立.
你发送的信息 2022-09-04 14:09:27
你好|4
服务端回应 2022-09-04 14:09:31
当前客户端不在线
你发送的信息 2022-09-04 14:09:42
你好
服务端回应 2022-09-04 14:09:42
你好
你发送的信息 2022-09-04 14:09:49
11
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

基于Netty实现WS

引入依赖

<dependency>
    <groupId>io.nettygroupId>
    <artifactId>netty-allartifactId>
    <version>4.1.39.Finalversion>
dependency>
<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-webartifactId>
dependency>
1
2
3
4
5
6
7
8
9

Netty配置

@Component
@Slf4j
public class NettyServer {
    /**
     * 默认8090
     */
    private int port = 8090;

    private EventLoopGroup mainGroup;
    private EventLoopGroup subGroup;
    private ServerBootstrap server;
    private ChannelFuture future;


    public NettyServer() {
        mainGroup = new NioEventLoopGroup();
        subGroup = new NioEventLoopGroup();
        server = new ServerBootstrap();
        server.option(ChannelOption.SO_BACKLOG, 1024);
        server.group(mainGroup, subGroup).channel(NioServerSocketChannel.class).localAddress(this.port).childHandler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                System.out.println("收到新连接:" + ch.localAddress());
                ch.pipeline().addLast(new HttpServerCodec());
                ch.pipeline().addLast(new ChunkedWriteHandler());
                ch.pipeline().addLast(new HttpObjectAggregator(8192));
                ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws", "WebSocket", true, 65536 * 10));
                ch.pipeline().addLast(new MyWebSocketHandler());
            }
        });
    }

    public void start() {
        this.future = server.bind(this.port);
        log.info("netty server 启动完毕,启动端口为：" + this.port);
    }
}
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

处理器

public class MyWebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    public static ChannelGroup channelGroup;
    static {
        channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
    }
    //客户端与服务器建立连接的时候触发，
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端建立连接，通道开启！");
        //添加到channelGroup通道组
        channelGroup.add(ctx.channel());
    }
    //客户端与服务器关闭连接的时候触发，
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("与客户端断开连接，通道关闭！");
        channelGroup.remove(ctx.channel());
    }
    //服务器接受客户端的数据信息，
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg){
        System.out.println("服务器收到的数据：" + msg.text());
        //sendMessage(ctx);
        sendAllMessage();
    }


    //给固定的人发消息
    private void sendMessage(ChannelHandlerContext ctx) {
        String message = "你好，"+ctx.channel().localAddress()+" 给固定的人发消息";
        ctx.channel().writeAndFlush(new TextWebSocketFrame(message));
    }

    //发送群消息,此时其他客户端也能收到群消息
    private void sendAllMessage(){
        String message = "我是服务器，这里发送的是群消息";
        channelGroup.writeAndFlush( new TextWebSocketFrame(message));
    }
}
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

启动类

@SpringBootApplication
public class Main implements CommandLineRunner {
    @Autowired
    private NettyServer nettyServer;

    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        this.nettyServer.start();
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

测试

这里推荐一个在线的测试工具：http://coolaf.com/zh/tool/chattest
输入地址：ws://localhost:8090/ws 就能愉快的测试了

基于tio实现WS(推荐)

tio是什么？

https://www.tiocloud.com/doc/tio/85
他的优势在于API设计易懂，尽量避免引入自创概念——最大限度降低学习成本。

引入依赖

<dependency>
    <groupId>org.t-iogroupId>
    <artifactId>tio-websocket-spring-boot-starterartifactId>
    <version>3.6.0.v20200315-RELEASEversion>
dependency>
<dependency>
    <groupId>org.springframework.bootgroupId>
    <artifactId>spring-boot-starter-webartifactId>
dependency>
1
2
3
4
5
6
7
8
9

编写配置

@Component
public class MyWebSocketMsgHandler implements IWsMsgHandler {
    @Override
    public HttpResponse handshake(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
        return httpResponse;
    }

    @Override
    public void onAfterHandshaked(HttpRequest httpRequest, HttpResponse httpResponse, ChannelContext channelContext) throws Exception {
        System.out.println("onAfterHandshaked  握手成功");
    }

    @Override
    public Object onBytes(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
        System.out.println("onBytes  接收到bytes消息");
        return null;
    }

    @Override
    public Object onClose(WsRequest wsRequest, byte[] bytes, ChannelContext channelContext) throws Exception {
        System.out.println("onClose");
        return null;
    }

    @Override
    public Object onText(WsRequest wsRequest, String s, ChannelContext channelContext) throws Exception {
        System.out.println("onText 接收到文本消息："+s);
        return "应答消息:"+s;
    }
}
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

主动推送

@RestController
@RequestMapping("/push")
public class PushController {

    @Autowired
    private TioWebSocketServerBootstrap bootstrap;

    /**
     * 消息群发
     * @param msg
     */
    @GetMapping("/msg")
    public void pushMessage(String msg){
        if (StrUtil.isEmpty(msg)){
            msg = "hello tio websocket spring boot starter";
        }
        Tio.sendToAll(bootstrap.getServerTioConfig(), WsResponse.fromText(msg,"utf-8"));
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

启动类

@SpringBootApplication
@EnableTioWebSocketServer
public class Main {
    public static void main(String[] args) {
        SpringApplication.run(Main.class, args);
    }
}
1
2
3
4
5
6
7

配置文件

tio:
  websocket:
    server:
      port: 9876
      heartbeat-timeout: 60000
1
2
3
4
5

测试

这里推荐一个在线的测试工具：http://coolaf.com/zh/tool/chattest
输入地址：ws://localhost:9876 就能愉快的测试了

基于tio实现websocket(集群版)

原理

引入Redis的发布订阅模式

demo代码已经发布到GitHub，需要请自取：https://github.com/shenhuan2021/websocket-cluster-demo