websocket框架netty和springboot的集成使用

maven依賴

<dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.33.Final</version>
        </dependency>

        <dependency>
            <groupId>cn.hutool</groupId>
            <artifactId>hutool-all</artifactId>
            <version>5.2.3</version>
        </dependency>

配置類

public class NettyConfig {
    /**
     * 定義一個channel組，管理所有的channel
     * GlobalEventExecutor.INSTANCE 是全局的事件執行器，是一個單例
     */
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 存放用戶與Chanel的對應信息，用於給指定用戶發送消息
     */
    private static ConcurrentHashMap<String, Channel> userChannelMap = new ConcurrentHashMap<>();

    private NettyConfig() {}

    /**
     * 獲取channel組
     * @return
     */
    public static ChannelGroup getChannelGroup() {
        return channelGroup;
    }

    /**
     * 獲取用戶channel map
     * @return
     */
    public static ConcurrentHashMap<String,Channel> getUserChannelMap(){
        return userChannelMap;
    }
}

服務類

@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:8090}")
    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));

                //針對客戶端，若10s內無讀事件則觸發心跳處理方法HeartBeatHandler#userEventTriggered
                ch.pipeline().addLast(new IdleStateHandler(10 , 0 , 0));
                //自定義空閑狀態檢測(自定義心跳檢測handler)
                ch.pipeline().addLast(new HeartBeatHandler());
                /*
                說明：
                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();

    }
}

處理類

@Component
@ChannelHandler.Sharable
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame> {

    private static final Logger log = LoggerFactory.getLogger(WebSocketHandler.class);

    /**
     * 一旦連接，第一個被執行
     * @param ctx
     * @throws Exception
     */
    @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());

        // 獲取用戶名
        JSONObject jsonObject = JSONUtil.parseObj(msg.text());
        String uid = jsonObject.getStr("uid");

        // 將用戶名作為自定義屬性加入到channel中，方便隨時channel中獲取用戶名
        AttributeKey<String> key = AttributeKey.valueOf("uid");
        ctx.channel().attr(key).setIfAbsent(uid);

        // 關聯channel
        NettyConfig.getUserChannelMap().put(uid,ctx.channel());

        // 回復消息
        ctx.channel().writeAndFlush(new TextWebSocketFrame("{\"code\":202,\"msg\":\"successConnect\"}"));
    }

    @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("uid");
        String userName = ctx.channel().attr(key).get();
        NettyConfig.getUserChannelMap().remove(userName);
    }
}

心跳包測試處理類

public class HeartBeatHandler extends ChannelInboundHandlerAdapter {

    private int lossConnectCount = 0;
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent){
            IdleStateEvent event = (IdleStateEvent)evt;
            if (event.state()== IdleState.READER_IDLE){
                lossConnectCount ++;
                if (lossConnectCount > 2){
                    ctx.channel().close();
                }
            }
        }else {
            super.userEventTriggered(ctx,evt);
        }
    }
}

消息對象的封裝

public class NettyPushMessageBody implements Serializable {

    private static final long serialVersionUID = 1L;

    private String uid;

    private String message;

    public String getUid() {
        return uid;
    }

    public void setUid(String uid) {
        this.uid = uid;
    }

    public String getMessage() {
        return message;
    }

    public void setMessage(String message) {
        this.message = message;
    }

    @Override
    public String toString() {
        return "NettyPushMessageBody{" +
                "uid='" + uid + '\'' +
                ", message='" + message + '\'' +
                '}';
    }
}

消息發送

@Component
public class MessageReceive {
    /**
     * 訂閱消息,發送給指定用戶
     * @param object
     */
    public void getMessageToOne(String object) {
        Jackson2JsonRedisSerializer serializer = getSerializer(NettyPushMessageBody.class);
        NettyPushMessageBody pushMessageBody = (NettyPushMessageBody) serializer.deserialize(object.getBytes());
        System.err.println("訂閱消息,發送給指定用戶：" + pushMessageBody.toString());

        // 推送消息
        String message = pushMessageBody.getMessage();
        String userId = pushMessageBody.getUid();
        ConcurrentHashMap<String, Channel> userChannelMap = NettyConfig.getUserChannelMap();
        Channel channel = userChannelMap.get(userId);
        if(!Objects.isNull(channel)){
            // 如果該用戶的客戶端是與本服務器建立的channel,直接推送消息
            channel.writeAndFlush(new TextWebSocketFrame(message));
        }
    }

    /**
     * 訂閱消息，發送給所有用戶
     * @param object
     */
    public void getMessageToAll(String object) {
        Jackson2JsonRedisSerializer serializer = getSerializer(String.class);
        String message = (String) serializer.deserialize(object.getBytes());
        System.err.println("訂閱消息，發送給所有用戶：" + message);
        NettyConfig.getChannelGroup().writeAndFlush(new TextWebSocketFrame(message));
    }

    private Jackson2JsonRedisSerializer getSerializer(Class clazz){
        //序列化對象（特別注意：發布的時候需要設置序列化；訂閱方也需要設置序列化）
        Jackson2JsonRedisSerializer seria = new Jackson2JsonRedisSerializer(clazz);
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL);
        seria.setObjectMapper(objectMapper);
        return seria;
    }
}

service服務層

public interface PushService {
    /**
     * 推送給指定用戶
     * @param userName
     * @param msg
     */
    void pushMsgToOne(String userName,String msg);

    /**
     * 推送給所有用戶
     * @param msg
     */
    void pushMsgToAll(String msg);
}

服務實現層

@Service
public class PushServiceImpl implements PushService {


    @Autowired
    private RedisTemplate redisTemplate;



    @Override
    public void pushMsgToOne(String uid, String msg){
        ConcurrentHashMap<String, Channel> userChannelMap = NettyConfig.getUserChannelMap();
        Channel channel = userChannelMap.get(uid);
        if(!Objects.isNull(channel)){
            // 如果該用戶的客戶端是與本服務器建立的channel,直接推送消息
            channel.writeAndFlush(new TextWebSocketFrame(msg));
        }else {
            // 發布，給其他服務器消費
            NettyPushMessageBody pushMessageBody = new NettyPushMessageBody();
            pushMessageBody.setUid(uid);
            pushMessageBody.setMessage(msg);
            redisTemplate.convertAndSend(Constants.PUSH_MESSAGE_TO_ONE,pushMessageBody);
        }

    }
    @Override
    public void pushMsgToAll(String msg){
        // 發布，給其他服務器消費
        redisTemplate.convertAndSend(Constants.PUSH_MESSAGE_TO_ALL,msg);
//        NettyConfig.getChannelGroup().writeAndFlush(new TextWebSocketFrame(msg));
    }
}