SpringBoot+WebSocket+Netty實現消息推送

一篇講一下Netty在項目中的應用場景之一：消息推送功能，可以滿足給所有用戶推送，也可以滿足給指定某一個用戶推送消息，創建的是SpringBoot項目，後臺服務端使用Netty技術，前端頁面使用WebSocket技術。

大概實現思路：

前端使用webSocket與服務端創建連接的時候，將用戶ID傳給服務端 服務端將用戶ID與channel關聯起來存儲，同時將channel放入到channel組中 如果需要給所有用戶發送消息，直接執行channel組的writeAndFlush()方法 如果需要給指定用戶發送消息，根據用戶ID查詢到對應的channel,然後執行writeAndFlush()方法 前端獲取到服務端推送的消息之後，將消息內容展示到文本域中

下面是具體的代碼實現，基本上每一步操作都配有註釋說明，配合註釋看應該還是比較容易理解的。

**第零步：**引入Netty的依賴，和一個工具包（只用到了json工具，可用其他json工具代替）

<code><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>/<code>

**第一步：**在NettyConfig中定義一個channel組，管理所有的channel，再定義一個map，管理用戶與channel的對應關係。

<code>package com.sixj.nettypush.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 sixiaojie
 * @date 2020-03-28-15:07
 */
public class NettyConfig {
    /**
     * 定義一個channel組，管理所有的channel
     * GlobalEventExecutor.INSTANCE 是全局的事件執行器，是一個單例
     */
    private static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

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

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

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

**第二步：**創建NettyServer，定義兩個EventLoopGroup，bossGroup輔助客戶端的tcp連接請求, workGroup負責與客戶端之前的讀寫操作，需要說明的是，需要開啟一個新的線程來執行netty server，要不然會阻塞主線程，到時候就無法調用項目的其他controller接口了。

<code>package com.sixj.nettypush.websocket;

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 sixiaojie
 * @date 2020-03-28-13:44
 */

@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();
    }
}/<socketchannel>/<code>

第三步： 具體實現業務的WebSocketHandler，具體實現邏輯看註釋

<code>package com.sixj.nettypush.websocket;

import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.sixj.nettypush.config.NettyConfig;
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;


/**
 * TextWebSocketFrame類型， 表示一個文本幀
 * @author sixiaojie
 * @date 2020-03-28-13:47
 */
@Component
@ChannelHandler.Sharable
public class WebSocketHandler extends SimpleChannelInboundHandler<textwebsocketframe> {
    private static final Logger log = LoggerFactory.getLogger(NettyServer.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());

        // 獲取用戶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);
    }
}/<string>/<string>/<textwebsocketframe>/<code>

**第四步：**具體消息推送的接口

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

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

接口實現類：

<code>import java.util.concurrent.ConcurrentHashMap;

/**
 * @author sixiaojie
 * @date 2020-03-30-20:10
 */
@Service
public class PushServiceImpl implements PushService {

    @Override
    public void pushMsgToOne(String userId, String msg){
        ConcurrentHashMap<string> 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));
    }
}/<string>/<code> 

controller：

<code>package com.sixj.nettypush.controller;

import com.sixj.nettypush.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 sixiaojie
 * @date 2020-03-30-20:08
 */
@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
     * @param msg
     */
    @PostMapping("/pushOne")
    public void pushMsgToOne(@RequestParam("userId") String userId,@RequestParam("msg") String msg){
        pushService.pushMsgToOne(userId,msg);
    }

}/<code>

**第五步：**前端html頁面

<code>

 

    
    <title>Title/<title>



    

/<code>

目前為止，所有代碼已經寫完了，測試一下

首先運行這個html文件，會看到服務端給前端返回的消息“服務器連接成功了！”，後端日誌會打印服務器收到消息：{"uid":"123456"}

然後使用postman測試推送的接口

作者：Sixj
鏈接：https://juejin.im/post/5e82ba62f265da47ac497e41

閱讀更多 妖精的雜七雜八 的文章

關鍵字: 消息 用戶 JSON