30分钟写一个聊天板

最近放假在家，无事学习了netty，写一个demo练手，快速编写一个简陋的聊天网页。

思路

基本的结构是后台采用netty，前端采用websocket和后台进行连接。

登陆：

• 前端用户发请求到netty服务器，服务器进行校验，返回响应

聊天：

• 前端用户将消息内容和聊天对象的ID以JSON报文的格式发给后台
• 后台经过Hadnler链拿到包，对里面的用户数据进行解析，并返回响应给用户前端
• 同时通过会话存储拿到聊天对象的channel并将消息发送给目标

本文阅读需要有对netty基础的了解，以及一点点前端websocket的知识

后台部分

创建服务器启动类：

package com.gdou.im.server;

import com.gdou.im.server.handler.WebSocketHandler;
import com.gdou.im.server.handler.LoginRequestHandler;
import com.gdou.im.server.handler.MessageRequestHandler;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
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.stream.ChunkedWriteHandler;

/**
 * @ProjectName: demo
 * @Package: com.gdou.im.server
 * @ClassName: NettyServer
 * @Author: carrymaniac
 * @Description: netty的服务器端
 * @Date: 2020/1/4 1:08 下午
 * @Version:
 */
public class NettyServer {

    public static void main(String[] args) throws InterruptedException {
        //定义线程组
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        final ServerBootstrap serverBootstrap = new ServerBootstrap();
        serverBootstrap.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<NioSocketChannel>(){
                    @Override
                    protected void initChannel(NioSocketChannel ch) throws Exception {
                        //Http编解码器，HttpServerCodec()
                        ch.pipeline().addLast(new HttpServerCodec());
                        //大数据流处理
                        ch.pipeline().addLast(new ChunkedWriteHandler());
                        //HttpObjectAggregator：聚合器，聚合了FullHTTPRequest、FullHTTPResponse。。。，当你不想去管一些HttpMessage的时候，直接把这个handler丢到管道中，让Netty自行处理即可
                        ch.pipeline().addLast(new HttpObjectAggregator(2048*64));
                        //WebSocketServerProtocolHandler：给客户端指定访问的路由（/ws），是服务器端处理的协议，当前的处理器处理一些繁重的复杂的东西，运行在一个WebSocket服务端
                        ch.pipeline().addLast(new WebSocketServerProtocolHandler("/ws"));
                        ch.pipeline().addLast(new WebSocketHandler());
                        ch.pipeline().addLast(new MessageRequestHandler());
                        ch.pipeline().addLast(new LoginRequestHandler());

                    }
                });
        ChannelFuture future = serverBootstrap.bind(8080).sync();
    }
}

其中，里面的WebSocketHandler、MessageRequestHandler、LoginRequestHandler是自定义的handler，下面分别展示：

WebSocketHandler

package com.gdou.im.server.handler;

import com.alibaba.fastjson.JSONObject;
import com.gdou.im.protocol.PacketCodeC;
import com.gdou.im.protocol.data.Data;
import com.gdou.im.protocol.data.request.LoginRequest;
import com.gdou.im.protocol.data.request.MessageRequest;
import com.gdou.im.protocol.Packet;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;

/**
 * @ProjectName: demo
 * @Package: com.gdou.im.server.handler
 * @ClassName: ChatHandler
 * @Author: carrymaniac
 * @Description: ChatHandler
 * @Date: 2020/1/28 12:01 上午
 * @Version:
 */
@Slf4j
public class WebSocketHandler extends SimpleChannelInboundHandler<TextWebSocketFrame>{

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TextWebSocketFrame msg) throws Exception {
        String text = msg.text();
        log.info("接收到了websocket包,内容为:{}",text);
        Packet packet = JSONObject.parseObject(text, Packet.class);
        if(packet!=null){
            Data decode = PacketCodeC.INSTANCE.decode(packet);
            //分发到下一个Handler
            if(decode instanceof MessageRequest){
                log.info("向下转型完成,内容为:{}",decode);
                ctx.fireChannelRead((MessageRequest)decode);
            }else if(decode instanceof LoginRequest){
                log.info("向下转型完成,内容为:{}",decode);
                ctx.fireChannelRead((LoginRequest)decode);
            }
        }
    }
}

WebSocketHandler主要职责是用于接收Handler链WebSocketServerProtocolHandler发下来的TextWebSocketFrame，解析其中的JSON正文为Packet (java对象)，然后转化为对应的每一个Request对象发送到Handler链的下一个Handler进行处理。

LoginRequestHandler

LoginRequestHandler主要用于处理WebSocketHandler发下来的MessageRequest数据，并生成LoginResponse响应将登陆情况发回给用户。

package com.gdou.im.server.handler;

import com.alibaba.fastjson.JSONObject;
import com.gdou.im.protocol.Packet;
import com.gdou.im.protocol.data.request.LoginRequest;
import com.gdou.im.protocol.data.response.LoginResponse;
import com.gdou.im.session.Session;
import com.gdou.im.util.SessionUtil;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;

import java.util.UUID;

import static com.gdou.im.protocol.command.Command.LOGIN_RESPONSE;

/**
 * @ProjectName: demo
 * @Package: com.gdou.im.server.handler
 * @ClassName: LoginRequestHandler
 * @Author: carrymaniac
 * @Description:
 * @Date: 2020/1/28 1:34 下午
 * @Version:
 */
@Slf4j
public class LoginRequestHandler extends SimpleChannelInboundHandler<LoginRequest> {

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LoginRequest msg) throws Exception {
        LoginResponse response = new LoginResponse();
        Packet packet = new Packet();
      //校验用户名和密码合法，这里没有实现，可以自行加入数据库校验等实现
        if(valid(msg)){
            //随机生成ID
            String userId = randomUserId();
            response.setSuccess(true);
            response.setUserId(userId);
            response.setUserName(msg.getUserName());
            //绑定session和channel，将用户信息和对应的channel进行绑定，以供之后使用
            SessionUtil.bindSession(new Session(userId,msg.getUserName()),ctx.channel());
            log.info("用户:{}登陆成功",msg.getUserName());
            //进行广播，对所有在线的成员channel发送一条消息
            SessionUtil.broadcast("用户: "+msg.getUserName()+"已上线,他的ID为: "+userId);
        }else {
            response.setReason("账号密码校验失败");
            response.setSuccess(false);
            log.info("用户:{}登陆失败",msg.getUserName());
        }
        packet.setData(JSONObject.toJSONString(response));
        packet.setCommand(LOGIN_RESPONSE);
      //将登陆成功的消息发给用户
        ctx.channel().writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(packet)));

    }


    /**
     * 进行登陆校验，todo 之后可以在这个方法中加入数据库进行校验
     * @param loginRequest
     * @return
     */
    private boolean valid(LoginRequest loginRequest) {
        return true;
    }

 		//生成一个用户ID
    private static String randomUserId() {
        return UUID.randomUUID().toString().split("-")[0];
    }

    /**
     * channel没有链接到远程节点的时候
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
      //当用户断开连接时，需要将其session和channel移除
        Session session = SessionUtil.getSession(ctx.channel());
        log.info("用户{}下线了，移除其session",session.getUserName());
        SessionUtil.unBindSession(ctx.channel());
        SessionUtil.broadcast("用户: "+session.getUserName()+"已下线");
    }

MessageRequestHandler

package com.gdou.im.server.handler;

import com.alibaba.fastjson.JSONObject;
import com.gdou.im.protocol.Packet;
import com.gdou.im.protocol.command.Command;
import com.gdou.im.protocol.data.Data;
import com.gdou.im.protocol.data.request.MessageRequest;
import com.gdou.im.protocol.data.response.MessageResponse;
import com.gdou.im.session.Session;
import com.gdou.im.util.SessionUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import lombok.extern.slf4j.Slf4j;

/**
 * @ProjectName: demo
 * @Package: com.gdou.im.server.handler
 * @ClassName: DataHandler
 * @Author: carrymaniac
 * @Description:
 * @Date: 2020/1/28 1:15 下午
 * @Version:
 */
@Slf4j
public class MessageRequestHandler extends SimpleChannelInboundHandler<MessageRequest> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, MessageRequest msg) throws Exception {
        log.info("拿到了数据，到达此处了:{}",msg);
        //通过channel获取到用户的信息
        Session session = SessionUtil.getSession(ctx.channel());
        //开始写回去
        Packet packetForConfirm = new Packet();
        packetForConfirm.setCommand(Command.MESSAGE_RESPONSE);
        MessageResponse responseForConfirm = new MessageResponse();
        responseForConfirm.setMessage(msg.getMessage());
        responseForConfirm.setFromUserName("你");
        packetForConfirm.setData(JSONObject.toJSONString(responseForConfirm));
        ctx.channel().writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(packetForConfirm)));

        //构建response,将消息发送给用户要发送的id用户
        //通过toId获取channel
        Channel channel = SessionUtil.getChannel(msg.getToId());
        if(channel!=null&& SessionUtil.hasLogin(channel)){
          //toID的用户在线，构建包发回给用户
            MessageResponse response = new MessageResponse();
            response.setFromUserId(session.getUserId());
            response.setFromUserName(session.getUserName());
            response.setMessage(msg.getMessage());
            Packet packetForToId = new Packet();
            packetForToId.setData(JSONObject.toJSONString(response));
            packetForToId.setCommand(Command.MESSAGE_RESPONSE);
            channel.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(packetForToId)));
        }else {
            log.info("用户并不在线");
        }
    }
}

到这，三个最重要的handler就介绍完成，还有一个比较重要的部分就是SessionUtil部分：

SessionUtil

//Session.java
@Data
@NoArgsConstructor
public class Session {
    // 用户唯一性标识
    private String userId;
    private String userName;

    public Session(String userId, String userName) {
        this.userId = userId;
        this.userName = userName;
    }

    @Override
    public String toString() {
        return userId + ":" + userName;
    }

}

//SessionUtil.java
package com.gdou.im.util;

import com.alibaba.fastjson.JSONObject;
import com.gdou.im.attribute.Attributes;
import com.gdou.im.protocol.Packet;
import com.gdou.im.protocol.data.response.MessageResponse;
import com.gdou.im.session.Session;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;

import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import static com.gdou.im.protocol.command.Command.SYSTEM_MESSAGE_RESPONSE;

public class SessionUtil {
  	//用于存储用户ID--->channel的对应关系
    private static final Map<String, Channel> userIdChannelMap = new ConcurrentHashMap<>();

    public static void bindSession(Session session, Channel channel) {
      //在map中存放绑定关系
        userIdChannelMap.put(session.getUserId(), channel);
      //在channel中存储用户信息
        channel.attr(Attributes.SESSION).set(session);
    }

    public static void unBindSession(Channel channel) {
        if (hasLogin(channel)) {
          //解除绑定
            userIdChannelMap.remove(getSession(channel).getUserId());
            channel.attr(Attributes.SESSION).set(null);
        }
    }

    public static boolean hasLogin(Channel channel) {

        return channel.hasAttr(Attributes.SESSION);
    }

    public static Session getSession(Channel channel) {

        return channel.attr(Attributes.SESSION).get();
    }

    public static Channel getChannel(String userId) {

        return userIdChannelMap.get(userId);
    }

  //广播
    public static void broadcast(String message){
        Packet packet = new Packet();
        packet.setCommand(SYSTEM_MESSAGE_RESPONSE);
        MessageResponse response = new MessageResponse();
        response.setMessage(message);
        response.setFromUserName("系统提醒");
        packet.setData(JSONObject.toJSONString(response));
        Set<Map.Entry<String, Channel>> entries = userIdChannelMap.entrySet();
        for(Map.Entry<String, Channel> entry :entries){
            Channel channel = entry.getValue();
            channel.writeAndFlush(new TextWebSocketFrame(JSONObject.toJSONString(packet)));
        }
    }
}

其余的后台代码比较简单，请参考我的github仓库的代码。

前端部分

前端部分因为本人没学过前端，因此写的比较稀烂，仅供参考：

<!DOCTYPE html>
<html>
	<head>
		<meta charset="utf-8" />
		<title>小聊天室DEMO</title>
	</head>
	<body>
		用户名:<input type="text" name="userName" id="userName"/><br>
		用户密码:<input type="password" name="userPassword" id="userPassword"/><br>
		<input type="button" onclick="CHAT.login()" name="loginButton" id="loginButton" value="登陆"/><br>
		<div id="">发送消息：</div>
		发送ID:<input type="text" name="toId" id="toId"/><br>
		发送内容:<input type="text" name="messageContent" id="messageContent"/><br>
        <input  type="button" onclick="CHAT.chat()" name="sendButton" id="sendButton" value="发送消息" />
		<hr>
		<div id="">接收消息列表：</div><br>
		<div id="receiveNsg" style="background-color: gainsboro;"></div>
		
		
		<script type="text/javascript">
			var user = {
				name:null,
				id:null
			}
			var COMMAND_CODE = {
					//登陆请求
					LOGIN_REQUEST:1,
					// 登陆消息响应
					LOGIN_RESPONSE:2,
					// 普通消息请求
					MESSAGE_REQUEST:3,
					// 普通消息响应
					MESSAGE_RESPONSE:4,
					// 系统消息响应
					SYSTEM_RESPONSE:-1
				},
			_this = this;
			window.CHAT = {
				socket: null,
				//初始化
				init: function(){
					//首先判断浏览器是否支持WebSocket
					if (window.WebSocket){
						that = this;
						CHAT.socket = new WebSocket("ws://localhost:8080/ws");
						CHAT.socket.onopen = function(){
							console.log("客户端与服务端建立连接成功");
						},
						CHAT.socket.onmessage = function(e){
							var receiveNsg = window.document.getElementById("receiveNsg");
							var html = receiveNsg.innerHTML;
							console.log("接收到消息："+e.data);
							var response = JSON.parse(e.data);
							// 说明是登陆的返回消息
							if(response.command==_this.COMMAND_CODE.LOGIN_RESPONSE){
								var result = JSON.parse(response.data);
								console.log(result);
								if(result.success==true){
									_this.user.name = result.userName;
									_this.user.id = result.userId;
									receiveNsg.innerHTML = html + "<br>" + 
									"用户登陆成功，您的ID为："+result.userId+",快去告诉你的朋友吧";
									return;
								}else{
									receiveNsg.innerHTML = html + "<br>" + 
									"用户登陆失败，原因是："+result.reason;
								}
							}else if(response.command==_this.COMMAND_CODE.MESSAGE_RESPONSE){
								var result = JSON.parse(response.data);
								receiveNsg.innerHTML = html + "<br>" + 
									"["+result.fromUserName+"]"+"说："+result.message;
								// 将ID设置到发送id框上去
								var toId = window.document.getElementById("toId");
								if(result.fromUserId!=_this.user.id){
									toId.value = result.fromUserId;
								}
								return;
							}else if(response.command==_this.COMMAND_CODE.SYSTEM_RESPONSE){
								var result = JSON.parse(response.data);
								receiveNsg.innerHTML = html + "<br>" + 
									"[系统提示] "+result.message;
								// 将ID设置到发送id框上去
								var toId = window.document.getElementById("toId");
								toId.value = result.fromUserId;
								return;
							}
						},
						CHAT.socket.onerror = function(){
							console.log("发生错误");
						},
						CHAT.socket.onclose = function(){
							console.log("客户端与服务端关闭连接成功");
						}		
					}else{
						alert("8102年都过了，升级下浏览器吧");
					}
				},
				chat: function(){
					var msg = window.document.getElementById("messageContent");
					var toId = window.document.getElementById("toId");
					var packet = {
						version:1,
						command:_this.COMMAND_CODE.MESSAGE_REQUEST,
						data:{
							fromid:_this.user.id,
							toid:toId.value,
							message:msg.value
						}
					}
					CHAT.socket.send(JSON.stringify(packet));
				},
				login:function(){
					var userName = window.document.getElementById("userName");
					var userPassword = window.document.getElementById("userPassword");
					var packet = {
						version:1,
						command:_this.COMMAND_CODE.LOGIN_REQUEST,
						data:{
							userName:userName.value,
							password:userPassword.value
						}
					}
					CHAT.socket.send(JSON.stringify(packet));
				}
			}
			CHAT.init();
		</script>
		
	</body>

</html>

大致效果如下图：

打开第二个标签页，再次登陆：

此时会发现，第一个标签页会出现提示：

在第一个标签页输入小红ID，以及内容，在第二个标签页显示如下：

小红用户接收到消息,并在发送ID框上填充上小明的ID。此时可以进行回复，小明用户效果图如下：

到此演示完毕，这个demo主要是为了自己记忆练习netty的主要用法。问题很多，大佬轻喷。

github地址