Netty实现一个简单聊天系统（点对点及服务端推送）

　　Netty是一个基于NIO，异步的，事件驱动的网络通信框架。由于使用Java提供 的NIO包中的API开发网络服务器代码量大，复杂，难保证稳定性。netty这类的网络框架应运而生。通过使用netty框架可以快速开发网络通信服务端，客户端。

　　本文主要通过一个简单的聊天程序来熟悉初步使用Nettyty进行简单服务端与客户端的开发。本聊天系统主要功能有点对点聊天及服务端推送消息。

程序结构：

　  Server端： IMServer 服务器启动类   ServerHandler 服务端核心类 负责客户端认证及消息转发

　  Client端：  IMClient 客户端启动类    ClientHandler 客户端核心类，负责客户端消息的发送及接收

　　Coder：MsgPackDecode和MsgPackEncode，负责消息的解码及编码实现，消息的编解码基于第三方库msgpack　

代码分析：

　  代码结构如下

　　　　

ApplicationContext类

功能比较简单，主要用来保存登录用户信息，以Map来存储，其中key为用户ID，value为客户端对应的ChannelHandlerContext对象。

import io.netty.channel.ChannelHandlerContext;
import java.util.HashMap;
import java.util.Map;

/**
 * Created by Andy on 2016/10/8.
 */
public class ApplicationContext {
    public static Map<Integer,ChannelHandlerContext> onlineUsers = new HashMap<Integer,ChannelHandlerContext>();
    public static void add(Integer uid,ChannelHandlerContext ctx){
        onlineUsers.put(uid,ctx);
    }

    public static void remove(Integer uid){
        onlineUsers.remove(uid);
    }

    public static ChannelHandlerContext getContext(Integer uid){
        return onlineUsers.get(uid);
    }
}

IMServerConfig接口

该接口主要用来存储服务端启动的配置信息，可改为配置文件实现

import com.wavemelody.nettyim.struts.MessageType;

/**
 * Created by Andy on 2016/10/8.
 */
public interface IMServerConfig {
    /**客户端配置*/
    int CLIENT_VERSION = 1;             //版本号
    /**服务端配置*/
    String SERVER_HOST = "127.0.0.1";   //服务器IP
    int  SERVER_PORT = 9090;            //服务器端口
    /**消息相关*/
    int SERVER_ID   = 0;                //表示服务器消息
    byte APP_IM = 1;                    //即时通信应用ID为1
    MessageType TYPE_MSG_CONNECT = MessageType.TYPE_AUTH;   //连接后第一次消息确认建立连接和发送认证信息
    MessageType TYPE_MSG_TEXT = MessageType.TYPE_TEXT;         //文本消息
    String MSG_DEFAULT = "";          //空消息
}

　　

ServerHandler

服务端主要的消息处理Handler，负责客户端认证之后，对客户端信息的保存及客户端点对点消息的转发以及程序异常时对资源的关闭等业务。

import com.wavemelody.nettyim.server.core.ApplicationContext;
import com.wavemelody.nettyim.struts.IMMessage;
import com.wavemelody.nettyim.struts.MessageType;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

/**
 * Created by Andy on 2016/10/8.
 */
public class ServerHandler extends ChannelInboundHandlerAdapter{
    private ChannelHandlerContext ctx;

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        System.out.println("服务端Handler创建。。。");
        super.handlerAdded(ctx);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("channelInactive");
        super.channelInactive(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        this.ctx = ctx;
        super.channelActive(ctx);
        System.out.println("有客户端连接：" + ctx.channel().remoteAddress().toString());
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        IMMessage message = (IMMessage)msg;
        if(message.getMsgType() == MessageType.TYPE_AUTH.value()){          //认证消息
            System.out.println("认证消息：" + msg);
            ApplicationContext.add(message.getUid(),ctx);
        }else if(message.getMsgType() == MessageType.TYPE_TEXT.value()){    //CHAT消息
            ChannelHandlerContext c = ApplicationContext.getContext(message.getReceiveId());
            if(c==null){           //接收方不在线，反馈给客户端
                message.setMsg("对方不在线！");
                ctx.writeAndFlush(message);
            }else{                 //将消转发给接收方
                System.out.println("转发消息：" + msg);
                c.writeAndFlush(message);
            }
        }

    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("与客户端断开连接:"+cause.getMessage());
        cause.printStackTrace();
        ctx.close();
    }

}

　　

IMServer

服务端的启动类，关于服务端，有以下几点需要说明

1. runServerCMD()方法用来启动控制台，启动后，可以对用户输入的内容进行消息推送。
2. MsgPackEncode和MsgPackDecode用于消息的编解码。使用的是MessagePack（API使用简单，编码后字节流特小，编解码速度较快，同时几乎支持所有主流编程语言，详情见官网：http://msgpack.org/）。这样我们可以随意编写实体用于发送消息，相关代码后边给出。
3. LengthFieldBasedFrameDecoder和LengthFieldPrepender：因为TCP底层传输数据时是不了解上层业务的，所以传输消息的时候很容易造成粘包/半包的情况（一条完整的消息被拆开或者完整或者不完整的多条消息被合并到一起发送、接收），这两个工具就是Netty提供的消息编码工具，2表示消息长度（不是正真的长度为2，是2个字节）。

import com.wavemelody.nettyim.codec.MsgPackDecode;
import com.wavemelody.nettyim.codec.MsgPackEncode;
import com.wavemelody.nettyim.server.config.IMServerConfig;
import com.wavemelody.nettyim.server.core.ApplicationContext;
import com.wavemelody.nettyim.server.handler.ServerHandler;
import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.io.IOException;
import java.util.Map;
import java.util.Scanner;

/**
 * Created by Andy on 2016/10/8.
 */
public class IMServer implements Runnable,IMServerConfig{
    public static void main(String[] args) throws IOException{
        new IMServer().start();
    }
    public void start() throws  IOException{
        new Thread(this).start();
        runServerCMD();
    }

    private IMMessage getMessage(){
        int toID = -1;
        IMMessage message = new IMMessage(
                APP_IM,
                CLIENT_VERSION,
                SERVER_ID,
                TYPE_MSG_TEXT.value(),
                toID,
                MSG_DEFAULT);
        return message;
    }

    private void runServerCMD()throws IOException{

        Scanner scanner = new Scanner(System.in);
        IMMessage message = null;
        do{
            message = getMessage();
            message.setMsg(scanner.nextLine());
        }while(sendMsg(message));
    }

    private boolean sendMsg(IMMessage msg){
        // 当用户输入quit表示退出，不在进行推送
        boolean result = msg.getMsg().equals("quit") ? false:true;
        if(result){
            int receiveID = msg.getReceiveId();
            String content = msg.getMsg();
            if(content.startsWith("#") && content.indexOf(":") != -1){
                try {
                    /**
                     * 用户输入指定的推送客户端
                     * 输入文本格式为： ＂＃8888:发送内容＂
                     * “#”和“：”之间内容为用户ID，“：”之后为推送消息内容
                     */
                    receiveID = Integer.valueOf(content.substring(1,content.indexOf(":")));
                    msg.setReceiveId(receiveID);
                    msg.setMsg(content.substring(content.indexOf(":")));
                } catch (NumberFormatException e) {
                    //解析失败则，默认发送所有
                    e.printStackTrace();
                }

            }

            /**
             * 默认推送所有用户（默认receiveID为-1）
             * */
            if(receiveID == -1){
                System.out.println("推送消息给所有在线用户：" + msg);
                for(Map.Entry<Integer,ChannelHandlerContext> entry: ApplicationContext.onlineUsers.entrySet()){
                    ChannelHandlerContext c = entry.getValue();
                    c.writeAndFlush(msg);
                }
            }else{
                ChannelHandlerContext ctx = ApplicationContext.getContext(receiveID);
                if(ctx!=null){
                    System.out.println("推送消息：" + msg);
                    ctx.writeAndFlush(msg);
                }

            }

        }
        return  result;
    }

    @Override
    public void run() {
        EventLoopGroup bossGroup  = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                    .option(ChannelOption.SO_BACKLOG,1024)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65536, 0, 2, 0, 2));
                            ch.pipeline().addLast("msgpack decoder",new MsgPackDecode());
                            ch.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
                            ch.pipeline().addLast("msgpack encoder",new MsgPackEncode());
                            ch.pipeline().addLast(new ServerHandler());
                        }
                    });
            ChannelFuture f = b.bind(SERVER_PORT).sync();
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

ClientHandler

客户端Handler

import com.wavemelody.nettyim.client.config.IMClientConfig;
import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import java.io.IOException;

/**
 * Created by Andy on 2016/10/8.
 */
public class ClientHandler extends ChannelInboundHandlerAdapter implements IMClientConfig{
    private ChannelHandlerContext ctx;

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("用户["+ UID + "]成功连接服务器");
        this.ctx = ctx;

        //通道建立时发送认证消息给服务器
        IMMessage message = new IMMessage(
                APP_IM,
                CLIENT_VERSION,
                UID,
                TYPE_MSG_AUTH.value(),
                DEFAULT_RECEIVE_ID,
                MSG_DEFAULT);
        sendMsg(message);
    }

    public boolean sendMsg(IMMessage msg) throws IOException {
        boolean result = msg.getMsg().equals("quit") ? false:true;
        if(result){
            if(msg.getMsgType() != MessageType.TYPE_AUTH.value()){
                System.out.println("认证消息： " + "client[" + msg.getUid() + "]:" + msg.getMsg());
            }
            //设置接收端ID和发送消息
            if(msg.getMsgType() == MessageType.TYPE_TEXT.value()){
                if(msg.getMsg().contains(":")){
                    String[] msgs = msg.getMsg().split(":");
                    String receiveIdStr =msgs[0].substring(1);
                    msg.setReceiveId(Integer.valueOf(receiveIdStr));
                    msg.setMsg(msgs[1]);
                }
            }
            ctx.writeAndFlush(msg);
        }
        return result;
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        IMMessage m = (IMMessage)msg;
        System.out.println("receive[" + m.getUid() + "]:" + m.getMsg());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println("与服务器断开连接：" + cause.getMessage());
        ctx.close();
    }
}

IMClient类

客户端启动类，同时启动控制台，将用户输入消息发送给指定的客户端

import com.wavemelody.nettyim.client.config.IMClientConfig;
import com.wavemelody.nettyim.client.handler.ClientHandler;
import com.wavemelody.nettyim.codec.MsgPackDecode;
import com.wavemelody.nettyim.codec.MsgPackEncode;
import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import java.io.IOException;
import java.util.Scanner;

/**
 * Created by Andy on 2016/10/8.
 */
public class IMClient implements Runnable,IMClientConfig{
    private ClientHandler clientHandler = new ClientHandler();
    public static void main(String[] args) throws IOException{
        new IMClient().start();
    }

    public void start() throws IOException{
        new Thread(this).start();
        runClientCMD();
    }
    public void runClientCMD() throws IOException{
        IMMessage message = new IMMessage(
                APP_IM,
                CLIENT_VERSION,
                UID,
                TYPE_MSG_TEXT.value(),
                DEFAULT_RECEIVE_ID,
                MSG_DEFAULT);
        Scanner scanner = new Scanner(System.in);
        do{
            message.setMsg(scanner.nextLine());
        }
        while (clientHandler.sendMsg(message));
    }

    @Override
    public void run() {
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(workerGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65536, 0, 2, 0, 2));
                    ch.pipeline().addLast("msgpack decoder",new MsgPackDecode());
                    ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2));
                    ch.pipeline().addLast("msgpack encoder",new MsgPackEncode());
                    ch.pipeline().addLast(clientHandler);
                }
            });
            ChannelFuture f = b.connect(SERVER_HOST, SERVER_PORT).sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            workerGroup.shutdownGracefully();
        }
    }
}

IMMessage类

该类即为通讯过程中的消息实体，即通讯协议，客户端进行消息发送，服务端进行消息推送时都需要将发送内容封装为IMMessage对象，才可以被识别。

import org.msgpack.annotation.Message;

/**
 * Created by Andy on 2016/10/8.
 */
@Message
public class IMMessage {
    //应用ID
    private byte appId;

    //版本
    private int version;

    //用户ID
    private int uid;

    //消息类型 0：登录 1：文字消息
    private byte msgType;

    //接收方
    private int receiveId;

    //消息内容
    private String msg;

    public IMMessage(){

    }

    /**
     * 构造方法
     * @param appId     应用通道
     * @param version   应用版本
     * @param uid       用户ID
     * @param msgType   消息类型
     * @param receiveId 消息接收者
     * @param msg       消息内容
     */
    public IMMessage(byte appId, int version, int uid, byte msgType, int receiveId, String msg) {
        this.appId = appId;
        this.version = version;
        this.uid = uid;
        this.msgType = msgType;
        this.receiveId = receiveId;
        this.msg = msg;
    }

    public byte getAppId() {
        return appId;
    }

    public void setAppId(byte appId) {
        this.appId = appId;
    }

    public int getVersion() {
        return version;
    }

    public void setVersion(int version) {
        this.version = version;
    }

    public int getUid() {
        return uid;
    }

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

    public byte getMsgType() {
        return msgType;
    }

    public void setMsgType(byte msgType) {
        this.msgType = msgType;
    }

    public int getReceiveId() {
        return receiveId;
    }

    public void setReceiveId(int receiveId) {
        this.receiveId = receiveId;
    }

    public String getMsg() {
        return msg;
    }

    public void setMsg(String msg) {
        this.msg = msg;
    }

    @Override
    public String toString() {
        return "IMMessage{" +
                "appId=" + appId +
                ", version=" + version +
                ", uid=" + uid +
                ", msgType=" + msgType +
                ", receiveId=" + receiveId +
                ", msg='" + msg + '\'' +
                '}';
    }
}

MessageType类

消息类型，通过枚举类型来约束消息中消息类型字段内容，防止出现系统不能识别的消息类型而发生异常。

/**
 * Created by Andy on 2016/10/9.
 */
public enum MessageType {
    TYPE_AUTH((byte)0),TYPE_LOGOUT((byte)1),TYPE_TEXT((byte)2),TYPE_EMPTY((byte)3);
    private byte value;
    MessageType(byte value){
        this.value = value;
    }
    public byte value(){
        return this.value;
    }
}

IMClientConfig接口

主要用来定义客户端启动配置信息常量，可改为配置文件实现方式。

import com.wavemelody.nettyim.struts.MessageType;

/**
 * Created by Andy on 2016/10/9.
 */
public interface IMClientConfig {
    /**客户端配置*/
    int CLIENT_VERSION = 1;         //版本号
    /**服务端配置*/
    String SERVER_HOST = "127.0.0.1";  //服务器IP
    int  SERVER_PORT = 9090;           //服务器端口
    /**消息相关*/
    byte APP_IM = 1;                //即时通信应用ID为1

    int UID = 8888;
    int DEFAULT_RECEIVE_ID = 9999;

    MessageType TYPE_MSG_AUTH = MessageType.TYPE_AUTH;         //连接后第一次消息确认建立连接和发送认证信息
    MessageType TYPE_MSG_TEXT = MessageType.TYPE_TEXT;         //文本消息
    String MSG_DEFAULT = "";          //默认为空消息
}

MsgPackEncode类

使用msgpack实现对消息的编码实现。

import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.msgpack.MessagePack;

/**
 * Created by Andy on 2016/10/8.
 */
public class MsgPackEncode extends MessageToByteEncoder<IMMessage> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, IMMessage msg, ByteBuf out) throws Exception {
        out.writeBytes(new MessagePack().write(msg));
    }
}

MsgPackDecode类

使用msgpack实现对消息的解码实现。

import com.wavemelody.nettyim.struts.IMMessage;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageDecoder;
import org.msgpack.MessagePack;

import java.util.List;

/**
 * Created by Andy on 2016/10/8.
 */
public class MsgPackDecode extends MessageToMessageDecoder<ByteBuf>{

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf msg, List<Object> out) throws Exception {
        final int length = msg.readableBytes();
        final  byte[] array = new byte[length];
        msg.getBytes(msg.readerIndex(),array,0,length);
        out.add(new MessagePack().read(array, IMMessage.class));
    }
}

　　通过以上代码基本上已经实现了一个简单的聊天程序，当然还存在很多地方需要优化。一个就是对TCP连接的优化，不是指定SO_KEEPALIVE属性，而是改为发送心跳消息来维持客户端和服务器的连接；然后就是链路中断后的重连实现，当出现中断之后由客户端等待一定时间重新发起连接操作，直至连接成功；另外一个就是重复登录验证，在客户端已经登录的情况下，要拒绝重复登录，防止客户端在异常状态下反复重连导致句柄资源被耗尽。