netty實現消息中心（二）基於netty搭建一個聊天室

前言

上篇博文（netty實現消息中心（一）思路整理
）大概說了下netty websocket消息中心的設計思路，這篇文章主要說說簡化版的netty聊天室代碼實現，支持群聊和點對點聊天。

此demo主要說明netty實現消息推送的基本使用方法，如果需要擴充其它功能，可以基於此腳手架擴展。
完整項目代碼地址：netty聊天室github源碼

介紹

1.登錄頁面

2.聊天頁面

核心代碼：

啟動netty服務，監聽端口

    private static void startNettyMsgServer() {
        // 使用多Reactor多線程模型，EventLoopGroup相當於線程池，內部維護一個或多個線程（EventLoop），每個EventLoop可處理多個Channel（單線程處理多個IO任務）
    	// 創建主線程組EventLoopGroup，專門負責建立連接
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 創建子線程組，專門負責IO任務的處理
        EventLoopGroup workGroup = new NioEventLoopGroup();
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workGroup);
            b.channel(NioServerSocketChannel.class);
            b.childHandler(new WebSocketChanneInitializer());
            System.out.println("服務端開啟等待客戶端連接....");
            Channel ch = b.bind(WebSocketConstant.WEB_SOCKET_PORT).sync().channel();

            //創建一個定長線程池，支持定時及周期性任務執行
            ScheduledExecutorService executorService = Executors.newScheduledThreadPool(3);
            WebSocketInfoService webSocketInfoService = new WebSocketInfoService();
            //定時任務:掃描所有的Channel，關閉失效的Channel
            executorService.scheduleAtFixedRate(webSocketInfoService::scanNotActiveChannel,
                    3, 60, TimeUnit.SECONDS);

            //定時任務:向所有客戶端發送Ping消息
            executorService.scheduleAtFixedRate(webSocketInfoService::sendPing,
                    3, 50, TimeUnit.SECONDS);

            ch.closeFuture().sync();


        } catch (Exception e) {
            e.printStackTrace();
        } finally {
//            //退出程序
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }
    }

netty ChannelHandler，負責處理通道的生命周期事件

package com.cola.chat_server.handler;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.UUID;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;
import com.cola.chat_server.constant.MessageCodeConstant;
import com.cola.chat_server.constant.MessageTypeConstant;
import com.cola.chat_server.constant.WebSocketConstant;
import com.cola.chat_server.model.WsMessage;
import com.cola.chat_server.service.WebSocketInfoService;
import com.cola.chat_server.util.DateUtils;
import com.cola.chat_server.util.NettyAttrUtil;
import com.cola.chat_server.util.RequestParamUtil;
import com.cola.chat_server.util.SessionHolder;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.DefaultFullHttpResponse;
import io.netty.handler.codec.http.FullHttpRequest;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.netty.handler.codec.http.HttpVersion;
import io.netty.handler.codec.http.websocketx.CloseWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PingWebSocketFrame;
import io.netty.handler.codec.http.websocketx.PongWebSocketFrame;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory;
import io.netty.util.CharsetUtil;



/**
 * Netty ChannelHandler，用來處理客戶端和服務端的會話生命周期事件（握手、建立連接、斷開連接、收消息等）
 * @Author 
 * @Description 接收請求，接收 WebSocket 信息的控制類
 */
public class WebSocketSimpleChannelInboundHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger logger = LoggerFactory.getLogger(WebSocketSimpleChannelInboundHandler.class);
    // WebSocket 握手工廠類
    private WebSocketServerHandshakerFactory factory = new WebSocketServerHandshakerFactory(WebSocketConstant.WEB_SOCKET_URL, null, false);
    private WebSocketServerHandshaker handshaker;
    private WebSocketInfoService websocketInfoService = new WebSocketInfoService();

    /**
     * 處理客戶端與服務端之間的 websocket 業務
     */
    private void handWebsocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        //判斷是否是關閉 websocket 的指令
        if (frame instanceof CloseWebSocketFrame) {
            //關閉握手
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            websocketInfoService.clearSession(ctx.channel());
            return;
        }
        //判斷是否是ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        // 判斷是否Pong消息
        if (frame instanceof PongWebSocketFrame) {
            ctx.writeAndFlush(new PongWebSocketFrame(frame.content().retain()));
            return;
        }
        //判斷是否是二進制消息，如果是二進制消息，拋出異常
        if (!(frame instanceof TextWebSocketFrame)) {
            System.out.println("目前我們不支持二進制消息");
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            throw new RuntimeException("【" + this.getClass().getName() + "】不支持消息");
        }
        // 獲取並解析客戶端向服務端發送的 json 消息
        String message = ((TextWebSocketFrame) frame).text();
        logger.info("消息：{}", message);
        JSONObject json = JSONObject.parseObject(message);
        try {
            String uuid = UUID.randomUUID().toString();
            String time = DateUtils.date2String(new Date(), "yyyy-MM-dd HH:mm:ss");
            json.put("id", uuid);
            json.put("sendTime", time);
            
            int code = json.getIntValue("code");
            switch (code) {
                //群聊
                case MessageCodeConstant.GROUP_CHAT_CODE:
                    //向連接上來的客戶端廣播消息
                	SessionHolder.channelGroup.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(json)));
                    break;
                //私聊
                case MessageCodeConstant.PRIVATE_CHAT_CODE:
                    //接收人id
                    String receiveUserId = json.getString("receiverUserId");
                    String sendUserId = json.getString("sendUserId");
                    String msg = JSONObject.toJSONString(json);
                    // 點對點挨個給接收人發送消息
                    for (Map.Entry<String, Channel> entry : SessionHolder.channelMap.entrySet()) {
                    	String userId = entry.getKey();
                    	Channel channel = entry.getValue();
                		if (receiveUserId.equals(userId)) {
                			channel.writeAndFlush(new TextWebSocketFrame(msg));
                		}
                    }
                    // 如果發給別人，給自己也發一條
                    if (!receiveUserId.equals(sendUserId)) {
                    	SessionHolder.channelMap.get(sendUserId).writeAndFlush(new TextWebSocketFrame(msg));
                    }
                    break;
                case MessageCodeConstant.SYSTEM_MESSAGE_CODE:
                	//向連接上來的客戶端廣播消息
                	SessionHolder.channelGroup.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(json)));
                	break;
                //pong
                case MessageCodeConstant.PONG_CHAT_CODE:
                    Channel channel = ctx.channel();
                    // 更新心跳時間
                    NettyAttrUtil.refreshLastHeartBeatTime(channel);
                default:
            }
        } catch(Exception e) {
            logger.error("轉發消息異常:", e);
            e.printStackTrace();
        }
    }

    /**
     * 客戶端與服務端創建連接的時候調用
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //創建新的 WebSocket 連接，保存當前 channel
        logger.info("————客戶端與服務端連接開啟————");
//        // 設置高水位
//        ctx.channel().config().setWriteBufferHighWaterMark();
//        // 設置低水位
//        ctx.channel().config().setWriteBufferLowWaterMark();
    }

    /**
     * 客戶端與服務端斷開連接的時候調用
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        logger.info("————客戶端與服務端連接斷開————");
        websocketInfoService.clearSession(ctx.channel());
    }

    /**
     * 服務端接收客戶端發送過來的數據結束之后調用
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    /**
     * 工程出現異常的時候調用
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        logger.error("異常:", cause);
        ctx.close();
    }

    /**
     * 服務端處理客戶端websocket請求的核心方法
     */
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, Object o) throws Exception {
        if (o instanceof FullHttpRequest) {
            //處理客戶端向服務端發起 http 請求的業務
            handHttpRequest(channelHandlerContext, (FullHttpRequest) o);
        } else if (o instanceof WebSocketFrame) {
            //處理客戶端與服務端之間的 websocket 業務
            handWebsocketFrame(channelHandlerContext, (WebSocketFrame) o);
        }
    }

    /**
     * 處理客戶端向服務端發起 http 握手請求的業務
     * WebSocket在建立握手時，數據是通過HTTP傳輸的。但是建立之后，在真正傳輸時候是不需要HTTP協議的。
     *
     * WebSocket 連接過程：
     * 首先，客戶端發起http請求，經過3次握手后，建立起TCP連接；http請求里存放WebSocket支持的版本號等信息，如：Upgrade、Connection、WebSocket-Version等；
     * 然后，服務器收到客戶端的握手請求后，同樣采用HTTP協議回饋數據；
     * 最后，客戶端收到連接成功的消息后，開始借助於TCP傳輸信道進行全雙工通信。
     */
    private void handHttpRequest(ChannelHandlerContext ctx, FullHttpRequest request) {
        // 如果請求失敗或者該請求不是客戶端向服務端發起的 http 請求，則響應錯誤信息
        if (!request.decoderResult().isSuccess()
                || !("websocket".equals(request.headers().get("Upgrade")))) {
            // code ：400
            sendHttpResponse(ctx, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }
        //新建一個握手
        handshaker = factory.newHandshaker(request);
        if (handshaker == null) {
            //如果為空，返回響應：不受支持的 websocket 版本
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            //否則，執行握手
            Map<String, String> params = RequestParamUtil.urlSplit(request.uri());
            String userId = params.get("userId");
            Channel channel = ctx.channel();
            NettyAttrUtil.setUserId(channel, userId);
            NettyAttrUtil.refreshLastHeartBeatTime(channel);
        	handshaker.handshake(ctx.channel(), request);
        	SessionHolder.channelGroup.add(ctx.channel());
        	SessionHolder.channelMap.put(userId, ctx.channel());
        	logger.info("握手成功，客戶端請求uri：{}", request.uri());
        	
        	// 推送用戶上線消息，更新客戶端在線用戶列表
        	Set<String> userList = SessionHolder.channelMap.keySet();
        	WsMessage msg = new WsMessage();
        	Map<String, Object> ext = new HashMap<String, Object>();
        	ext.put("userList", userList);
        	msg.setExt(ext);
        	msg.setCode(MessageCodeConstant.SYSTEM_MESSAGE_CODE);
        	msg.setType(MessageTypeConstant.UPDATE_USERLIST_SYSTEM_MESSGAE);
        	SessionHolder.channelGroup.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(msg)));
        	
        }
    }


    /**
     * 服務端向客戶端響應消息
     */
    private void sendHttpResponse(ChannelHandlerContext ctx, DefaultFullHttpResponse response) {
        if (response.status().code() != 200) {
            //創建源緩沖區
            ByteBuf byteBuf = Unpooled.copiedBuffer(response.status().toString(), CharsetUtil.UTF_8);
            //將源緩沖區的數據傳送到此緩沖區
            response.content().writeBytes(byteBuf);
            //釋放源緩沖區
            byteBuf.release();
        }
        //寫入請求，服務端向客戶端發送數據
        ChannelFuture channelFuture = ctx.channel().writeAndFlush(response);
        if (response.status().code() != 200) {
        	/**
        	 * 如果請求失敗，關閉 ChannelFuture
        	 * ChannelFutureListener.CLOSE 源碼：future.channel().close();
        	 */
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }
    }
}

會話工具類，保存用戶和通道的對應關系，用於廣播和點對點聊天

/**
 * netty會話管理
 * @author 
 *
 */
public class SessionHolder {
	
    /**
     * 存儲每個客戶端接入進來時的 channel 對象
     * 主要用於使用 writeAndFlush 方法廣播信息
     */
    public static ChannelGroup channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 用於客戶端和服務端握手時存儲用戶id和netty Channel對應關系
     */
    public static Map<String, Channel> channelMap = new ConcurrentHashMap<String, Channel>(); 

}

主要代碼就是以上部分，如果需要擴充其它功能，可以基於此腳手架擴展。
完整項目代碼地址：netty聊天室github源碼

此demo主要用於展示netty實現消息推送的基本使用方法，用於生產還存在以下單機問題：
1.無法支撐過高連接數
2.廣播時帶寬有限
3.不能實現高可用
4.無法橫向擴展
后期將集成zookeeper，做一版netty集群的聊天室。