前言

上篇博文(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 {

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 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 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 userList = SessionHolder.channelMap.keySet();

WsMessage msg = new WsMessage();

Map ext = new HashMap();

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 channelMap = new ConcurrentHashMap();

}

主要代碼就是以上部分，如果需要擴充其它功能，可以基于此腳手架擴展。

完整項目代碼地址：netty聊天室github源碼

此demo主要用于展示netty實現消息推送的基本使用方法，用于生產還存在以下單機問題：

1.無法支撐過高連接數

2.廣播時帶寬有限

3.不能實現高可用

4.無法橫向擴展

后期將集成zookeeper，做一版netty集群的聊天室。

總結

以上是生活随笔為你收集整理的java netty聊天室_netty实现消息中心（二）基于netty搭建一个聊天室的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。