SpringBoot整合Netty+WebSocket

構建環境

pom.xml

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.4.5</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.cyz</groupId>
    <artifactId>websocket-netty</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>websocket-netty</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
            <version>2.4.2</version>
        </dependency>
​
​
<!--       需要依賴-->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.63.Final</version>
        </dependency>
        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.6.4</version>
        </dependency>
        <!--       需要依賴-->
​
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-thymeleaf</artifactId>
        </dependency>
​
        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
    </dependencies>
​
    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <excludes>
                        <exclude>
                            <groupId>org.projectlombok</groupId>
                            <artifactId>lombok</artifactId>
                        </exclude>
                    </excludes>
                </configuration>
            </plugin>
        </plugins>
    </build>
​
</project>
​

package com.cyz.websocketnetty.controller;
​
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
​
@RestController
public class HelloController {
    @GetMapping("/hello")
    public String sayHello(@RequestParam String name) {
        return "hello：" + name;
   }
}

實現過程

配置文件

server.port=9090
spring.mvc.view.prefix=/
spring.mvc.view.suffix=.html

NettyConfig

package com.cyz.websocketnetty.config;
​
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
​
import java.util.concurrent.ConcurrentHashMap;
​
/**
 * @author lian-chen
 * 配置類
 */
public class NettyConfig {
    /**
     * 定義一個channel組，管理所有的channel
     * GlobalEventExecutor.INSTANCE 是全局的事件執行器，是一個單例
     */
    private static final ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
​
    /**
     * 存放用戶與Chanel的對應信息，用於給指定用戶發送消息
     */
    private static final ConcurrentHashMap<String, Channel> USER_CHANNEL_MAP = new ConcurrentHashMap<>();
​
    private NettyConfig() {
   }
​
    /**
     * 獲取channel組
     */
    public static ChannelGroup getChannelGroup() {
        return CHANNEL_GROUP;
   }
​
    /**
     * 獲取用戶channel map
     */
    public static ConcurrentHashMap<String, Channel> getUserChannelMap() {
        return USER_CHANNEL_MAP;
   }
}

NettyServer

package com.cyz.websocketnetty.service;
​
import com.cyz.websocketnetty.handle.WebSocketHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.HttpServerCodec;
import io.netty.handler.codec.http.websocketx.WebSocketServerProtocolHandler;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.stream.ChunkedWriteHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
​
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.net.InetSocketAddress;
​
/**
 * @author lian-chen
 */
@Component
public class NettyServer {
    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);
    /**
     * webSocket協議名
     */
    private static final String WEBSOCKET_PROTOCOL = "WebSocket";
​
    /**
     * 端口號
     */
    @Value("${webSocket.netty.port:58080}")
    private int port;
​
    /**
     * webSocket路徑
     */
    @Value("${webSocket.netty.path:/webSocket}")
    private String webSocketPath;
​
    @Autowired
    private WebSocketHandler webSocketHandler;
​
    private EventLoopGroup bossGroup;
​
    private EventLoopGroup workGroup;
​
    /**
     * 啟動
     *
     * @throws InterruptedException
     */
    private void start() throws InterruptedException {
        bossGroup = new NioEventLoopGroup();
        workGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        // bossGroup輔助客戶端的tcp連接請求, workGroup負責與客戶端之前的讀寫操作
        bootstrap.group(bossGroup, workGroup);
        // 設置NIO類型的channel
        bootstrap.channel(NioServerSocketChannel.class);
        // 設置監聽端口
        bootstrap.localAddress(new InetSocketAddress(port));
        // 連接到達時會創建一個通道
        bootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
​
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                // 流水線管理通道中的處理程序（Handler），用來處理業務
                // webSocket協議本身是基於http協議的，所以這邊也要使用http編解碼器
                ch.pipeline().addLast(new HttpServerCodec());
                ch.pipeline().addLast(new ObjectEncoder());
                // 以塊的方式來寫的處理器
                ch.pipeline().addLast(new ChunkedWriteHandler());
        /*
        說明：
        1、http數據在傳輸過程中是分段的，HttpObjectAggregator可以將多個段聚合
        2、這就是為什么，當瀏覽器發送大量數據時，就會發送多次http請求
         */
                ch.pipeline().addLast(new HttpObjectAggregator(8192));
        /*
        說明：
        1、對應webSocket，它的數據是以幀（frame）的形式傳遞
        2、瀏覽器請求時 ws://localhost:58080/xxx 表示請求的uri
        3、核心功能是將http協議升級為ws協議，保持長連接
        */
                ch.pipeline().addLast(new WebSocketServerProtocolHandler(webSocketPath, WEBSOCKET_PROTOCOL, true, 65536 * 10));
                // 自定義的handler，處理業務邏輯
                ch.pipeline().addLast(webSocketHandler);
​
           }
       });
        // 配置完成，開始綁定server，通過調用sync同步方法阻塞直到綁定成功
        ChannelFuture channelFuture = bootstrap.bind().sync();
        log.info("Server started and listen on:{}", channelFuture.channel().localAddress());
        // 對關閉通道進行監聽
        channelFuture.channel().closeFuture().sync();
   }
​
    /**
     * 釋放資源
     *
     * @throws InterruptedException
     */
    @PreDestroy
    public void destroy() throws InterruptedException {
        if (bossGroup != null) {
            bossGroup.shutdownGracefully().sync();
       }
        if (workGroup != null) {
            workGroup.shutdownGracefully().sync();
       }
   }
​
    @PostConstruct()
    public void init() {
        //需要開啟一個新的線程來執行netty server 服務器
        new Thread(() -> {
            try {
                start();
           } catch (InterruptedException e) {
                e.printStackTrace();
           }
       }).start();
   }
}

WebSocketHandler

package com.cyz.websocketnetty.handle;
​
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.cyz.websocketnetty.config.NettyConfig;
import com.cyz.websocketnetty.service.NettyServer;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
​
/**
 * @author lian-chen
 */
@Component
@ChannelHandler.Sharable
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {
    private static final Logger log = LoggerFactory.getLogger(NettyServer.class);
​
    /**
     * @param ctx   &#x4E0A;&#x4E0B;&#x6587;
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("handlerAdded 被調用"+ctx.channel().id().asLongText());
        // 添加到channelGroup 通道組
        NettyConfig.getChannelGroup().add(ctx.channel());
   }
​
    /**
     * 讀取數據
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        log.info("服務器收到消息：{}",msg.text());
​
        // 獲取用戶ID,關聯channel
        JSONObject jsonObject = JSONUtil.parseObj(msg.text());
        String uid = jsonObject.getStr("uid");
        NettyConfig.getUserChannelMap().put(uid,ctx.channel());
​
        // 將用戶ID作為自定義屬性加入到channel中，方便隨時channel中獲取用戶ID
        AttributeKey<String> key = AttributeKey.valueOf("userId");
        ctx.channel().attr(key).setIfAbsent(uid);
​
        // 回復消息
        ctx.channel().writeAndFlush(new TextWebSocketFrame("服務器連接成功！"));
   }
​
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        log.info("handlerRemoved 被調用"+ctx.channel().id().asLongText());
        // 刪除通道
        NettyConfig.getChannelGroup().remove(ctx.channel());
        removeUserId(ctx);
   }
​
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.info("異常：{}",cause.getMessage());
        // 刪除通道
        NettyConfig.getChannelGroup().remove(ctx.channel());
        removeUserId(ctx);
        ctx.close();
   }
​
    /**
     * 刪除用戶與channel的對應關系
     * @param ctx   上下文
     */
    private void removeUserId(ChannelHandlerContext ctx){
        AttributeKey<String> key = AttributeKey.valueOf("userId");
        String userId = ctx.channel().attr(key).get();
        NettyConfig.getUserChannelMap().remove(userId);
   }
}

PushService

package com.cyz.websocketnetty.service;
​
​
/**
 * @author lian-chen
 */
public interface PushService {
    /**
     * 推送給指定用戶
     *
     * @param userId 用戶ID
     * @param msg   消息
     */
    void pushMsgToOne(String userId, String msg);
​
    /**
     * 推送給所有用戶
     *
     * @param msg 消息
     */
    void pushMsgToAll(String msg);
}

PushServiceImpl

package com.cyz.websocketnetty.service.impl;
​
import com.cyz.websocketnetty.config.NettyConfig;
import com.cyz.websocketnetty.service.PushService;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import org.springframework.stereotype.Service;
import io.netty.channel.Channel;
​
import java.util.concurrent.ConcurrentHashMap;
​
/**
 * @author lian-chen
 */
@Service
public class PushServiceImpl implements PushService {
​
    @Override
    public void pushMsgToOne(String userId, String msg){
        ConcurrentHashMap<String, Channel> userChannelMap = NettyConfig.getUserChannelMap();
        Channel channel = userChannelMap.get(userId);
        channel.writeAndFlush(new TextWebSocketFrame(msg));
   }
    @Override
    public void pushMsgToAll(String msg){
        NettyConfig.getChannelGroup().writeAndFlush(new TextWebSocketFrame(msg));
   }
}

PushController

package com.cyz.websocketnetty.controller;
​
import com.cyz.websocketnetty.service.PushService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
​
/**
 * @author lian-chen
 */
@RestController
@RequestMapping("/push")
public class PushController {
​
    @Autowired
    private PushService pushService;
​
    /**
     * 推送給所有用戶
     *
     * @param msg 消息
     */
    @PostMapping("/pushAll")
    public void pushToAll(@RequestParam("msg") String msg) {
        pushService.pushMsgToAll(msg);
   }
​
    /**
     * 推送給指定用戶
     *
     * @param userId 用戶ID
     * @param msg   消息
     */
    @PostMapping("/pushOne")
    public void pushMsgToOne(@RequestParam("userId") String userId, @RequestParam("msg") String msg) {
        pushService.pushMsgToOne(userId, msg);
   }
}

IndexController

package com.cyz.websocketnetty.controller;
​
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
​
/**
 * @author lian-chen
 */
@Controller
public class IndexController {
​
    @GetMapping("/socket/{userId}")
    public String testHtml(@PathVariable("userId") String userId, Model model) {
        model.addAttribute("userId",userId);
        return "html/testwebsocket";
   }
}

testwebsocket.html

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>Title</title>
</head>
<body>
<script>
    var socket;
    // 判斷當前瀏覽器是否支持webSocket
    if (window.WebSocket) {
        socket = new WebSocket("ws://192.168.10.47:58080/webSocket")
        // 相當於channel的read事件，ev 收到服務器回送的消息
        socket.onmessage = function (ev) {
            var rt = document.getElementById("responseText");
            rt.value = rt.value + "\n" + ev.data;
       }
        // 相當於連接開啟
        socket.onopen = function (ev) {
            let userId = [[${userId}]];
            var rt = document.getElementById("responseText");
            rt.value = "連接開啟了..."
            socket.send(
                JSON.stringify({
                    // 連接成功將，用戶ID傳給服務端
                    uid: userId
               })
           );
       }
        // 相當於連接關閉
        socket.onclose = function (ev) {
            var rt = document.getElementById("responseText");
            rt.value = rt.value + "\n" + "連接關閉了...";
       }
   } else {
        alert("當前瀏覽器不支持webSocket")
   }
​
​
</script>
<form onsubmit="return false">
    <textarea id="responseText" style="height: 150px; width: 300px;"></textarea>
    <input type="button" value="清空內容" onclick="document.getElementById('responseText').value=''">
</form>
</body>
</html>

訪問測試

• http://localhost:9090/socket/1

• http://localhost:9090/socket/2

• http://localhost:9090/socket/3