前言

像使用linux系统一样通过一个cmd窗口和朋友进行聊天是一件很炫酷的事情，同时也是规避公司聊天记录审查的一种手段，尤其一些公司禁止使用微信等聊天工具，这给日常同事们私下沟通交流增加了一些不便。

最终效果

使用netty实现自定义通信协议

1、定义消息加解密类
加密类

package pers.cz.netty.protocal;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayOutputStream;
import java.io.ObjectOutputStream;
import java.nio.charset.Charset;

/**
 * @program: PostGirl-panent
 * @description: 消息加密
 * @author: Cheng Zhi
 * @create: 2022-10-04 15:31
 **/
public class TcpMessageEncoder extends MessageToByteEncoder<Message> {

    private static final Logger log = LoggerFactory.getLogger(TcpMessageEncoder.class);

    /*
    +------------------------------------------------------------------------------------------------------------------------+
    | 魔数 8byte 使用String描述   |     报文类型 1byte      | attachments附件信息（长度不定）  |  数据长度 8byte   使用long描述      |
    +------------------------------------------------------------------------------------------------------------------------+
    |                   数据内容 （长度不定）                                                    |
    +-------------------------------------------------------------------------------------------+
    */
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf byteBuf) throws Exception {

        // 写入魔数
        byteBuf.writeBytes(ProtocalConst.POCKET_MAGIC_NUM.getBytes());
        // 写入消息类型
        byteBuf.writeByte(message.getMsgType().type());
        // 写入附件信息
        byteBuf.writeShort(message.getAttachments().size());	// 写入当前消息的附加参数数量
        message.getAttachments().forEach((key, value) -> {
            Charset charset = Charset.defaultCharset();
            byteBuf.writeInt(key.length());	// 写入键的长度
            byteBuf.writeCharSequence(key, charset);	// 写入键数据
            byteBuf.writeInt(value.length());	// 写入值的长度
            byteBuf.writeCharSequence(value, charset);	// 写入值数据
        });
        // 写入消息长度和数据内容
        final Object content = message.getContent();
        if (content == null) {
            byteBuf.writeLong(0L);
        } else {
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(bos);
            oos.writeObject(content);
            byte[] bytes = bos.toByteArray();
            byteBuf.writeLong(bytes.length);
            // 主体消息
            byteBuf.writeBytes(bytes);
        }

    }
}

加密类

package pers.cz.netty.protocal;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.ByteArrayInputStream;
import java.io.ObjectInputStream;
import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 * @program: PostGirl-panent
 * @description: 消息解密
 * @author: Cheng Zhi
 * @create: 2022-10-04 15:30
 **/
public class TcpMessageDecoder extends ByteToMessageDecoder {

    private static final Logger log = LoggerFactory.getLogger(TcpMessageDecoder.class);

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {

        // 解密魔数
        byte[] bytes = new byte[8];
        byteBuf.readBytes(bytes);
        String magicNum = new String(bytes, StandardCharsets.UTF_8);

        if (!ProtocalConst.POCKET_MAGIC_NUM.equals(magicNum)) {
            log.error("未知数据包，magicNum=" + magicNum );
            return;
        }

        // 解密消息类型
        byte msgType = byteBuf.readByte();

        final MsgType type = MsgType.getMsgType(msgType);
        if (type == null) {
            log.error("未知操作类型：" + msgType);
            return;
        }

        Message msg = new Message();
        msg.setMsgType(type);

        // 解密消息附件
        short attachmentSize = byteBuf.readShort();	// 读取附件长度
        for (short i = 0; i < attachmentSize; i++) {
            int keyLength = byteBuf.readInt();	// 读取键长度和数据
            CharSequence key = byteBuf.readCharSequence(keyLength, Charset.defaultCharset());
            int valueLength = byteBuf.readInt();	// 读取值长度和数据
            CharSequence value = byteBuf.readCharSequence(valueLength, Charset.defaultCharset());
            msg.addAttachment(key.toString(), value.toString());
        }


        // 读取消息长度
        long contentLenth = byteBuf.readLong();

        msg.setContentLength(contentLenth);
        if(contentLenth > 0){
            // 读取主体内容
            byte[] contents = new byte[(int) contentLenth];
            byteBuf.readBytes(contents,0, (int) contentLenth);
            ByteArrayInputStream bis=new ByteArrayInputStream(contents);
            ObjectInputStream ois=new ObjectInputStream(bis);
            msg.setContent(ois.readObject());
        }

        list.add(msg);
    }
}

服务端：

package pers.cz.netty;

import io.netty.bootstrap.ServerBootstrap;
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.NioServerSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.codec.LengthFieldPrepender;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import pers.cz.jefconfig.config.JefConfiguration;
import pers.cz.netty.config.ServerNettyConfigItem;
import pers.cz.netty.handle.ServerHeartBeatHandler;
import pers.cz.netty.handle.ServerIdleStateHandler;
import pers.cz.netty.handle.ServerMessageHandler;
import pers.cz.netty.protocal.ProtocalConst;
import pers.cz.netty.protocal.TcpMessageDecoder;
import pers.cz.netty.protocal.TcpMessageEncoder;

/**
 * @program: PostGirl-panent
 * @description: 服务端
 * @author: Cheng Zhi
 * @create: 2022-10-05 08:08
 **/
public class TcpNettyServer {

    private static final Logger log = LoggerFactory.getLogger(TcpNettyServer.class);

    private int port;

    public TcpNettyServer(int port) {
        this.port = port;
    }

    public void start(ServerMessageHandler serverMessageHandler) {

        showBanner();
        // 负责连接
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        // 负责除连接之外的工作
        EventLoopGroup workerGroup = new NioEventLoopGroup(2);
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)

                    //实例化ServerSocketChannel
                    .channel(NioServerSocketChannel.class)

                    //设置ServerSocketChannel的TCP参数
                    .option(ChannelOption.SO_BACKLOG, 1024)

                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // http服务端编解码规则
                            ch.pipeline().addLast("idleStateHandler", new IdleStateHandler(JefConfiguration.getInt(ServerNettyConfigItem.SERVER_HEART_READ_TIME, 15), 0, 0));
                            // 添加超时检测
                            ch.pipeline().addLast("ServerIdleStateHandler", new ServerIdleStateHandler());
                            // 长度解码：包最大值 数据内容偏移量
                            ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(ProtocalConst.PACKAGE_MAX_SIZE,0,8,0,8));
                            ch.pipeline().addLast(new LengthFieldPrepender(8));
                            ch.pipeline().addLast(new TcpMessageEncoder());
                            ch.pipeline().addLast(new TcpMessageDecoder());
                            ch.pipeline().addLast(new ServerHeartBeatHandler());
                            ch.pipeline().addLast(serverMessageHandler);

                        }
                    });
            //绑定监听端口，调用sync同步阻塞方法等待绑定操作完
            ChannelFuture future = b.bind(this.port).sync();
            final boolean success = future.isSuccess();
            log.debug("netty service has been started, listen on: {} port", this.port);
            //成功绑定到端口之后,给channel增加一个管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程。
            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("netty service failed to start on " + port + " port", e);
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    public void showBanner() {
        log.info("================================================================================");
        log.info("==     ~~~   jef TCP netty server is starting    ~~~                         ==");
        log.info("================================================================================");
    }
/*    public static void main(String[] args) {
        new NettyServer(8881).start(null);
    }*/
}

客户端：

package pers.cz.netty.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
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 pers.cz.netty.filter.IFilter;
import pers.cz.netty.handler.HeartBeatHandler;
import pers.cz.netty.handler.IdleStateHandler;
import pers.cz.netty.handler.MessageHandler;
import pers.cz.netty.listener.Listener;
import pers.cz.netty.protocal.Message;
import pers.cz.netty.protocal.ProtocalConst;
import pers.cz.netty.protocal.TcpMessageDecoder;
import pers.cz.netty.protocal.TcpMessageEncoder;

import java.util.ArrayList;
import java.util.List;
import java.util.logging.Logger;

/**
 * @program: PostGirl-panent
 * @description: 客户端触发类
 * @author: Cheng Zhi
 * @create: 2022-10-04 15:03
 **/
public class TcpNettyClient implements NettyClient {

    /**
     * 为了后续集成agent时不产生循环打印自己的日志。
     */
    private static final Logger log = Logger.getLogger(TcpNettyClient.class.getSimpleName());

    private String hosts;

    private Integer port;

    private Bootstrap bootstrap;

    private static Channel channel;

    private HeartBeatHandler heartBeatHandler = new HeartBeatHandler(this);

    private List<Listener> messageListener = new ArrayList<>();

    private MessageHandler messageHandler;

    private IFilter<Message, Channel> filter;

    EventLoopGroup group = new NioEventLoopGroup();

    @Deprecated
    public TcpNettyClient(String hosts, Integer port) {
        this(hosts, port, null);
    }

    public TcpNettyClient(String hosts, Integer port, MessageHandler messageHandler) {
        this.messageHandler = messageHandler;
        showBanner();
        this.hosts = hosts;
        this.port = port;
        init();
    }

    /**
     * 初始化
     */
    private void init() {

        if (messageHandler == null) {
            throw new RuntimeException("请指定消息处理器！");
        }
        bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup())
                .channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)
                .handler(new ChannelInitializer<SocketChannel>(){

                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline().addLast(new LengthFieldBasedFrameDecoder(ProtocalConst.PACKAGE_MAX_SIZE, 0, 8, 0, 8));
                        socketChannel.pipeline().addLast(new LengthFieldPrepender(8));
                        socketChannel.pipeline().addLast(new TcpMessageDecoder());
                        socketChannel.pipeline().addLast(new TcpMessageEncoder());
                        socketChannel.pipeline().addLast(heartBeatHandler);
                        socketChannel.pipeline().addLast(messageHandler);
                        socketChannel.pipeline().addLast(new IdleStateHandler());
                    }
                });

    }

    @Override
    public void setFilter(IFilter filter) {
        this.filter = filter;
    }

    public ChannelFuture connect() {

        synchronized (this) {
            final ChannelFuture f = bootstrap.connect(this.hosts, this.port);
            f.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    if (future.isSuccess()) {
                        channel = future.channel();
                    } else {
                        future.channel().pipeline().fireChannelInactive();
                    }
                }
            });
            return f;
        }
    }

    public void shutDown() {
        if (group != null) {
            group.shutdownGracefully();
        }
    }

    public boolean send(Message message) {

        if (channel == null || !channel.isActive()) {
            log.info("未连接服务器，发送失败");
            return false;
        }

        if (filter != null && filter.doFilter(message, channel)) {
            return false;
        }
        for (Listener listener : messageListener) {
            listener.before(message);
        }
        ChannelFuture f = channel.writeAndFlush(message);
        boolean success = f.isSuccess();
        if (f.cause() != null) {
            System.out.println(f.cause().getCause().getMessage());
        }
        for (Listener listener : messageListener) {
            listener.after(message);
        }
        return success;
    }

    public static boolean sendMsg(Message message) {
        if (channel == null || !channel.isActive()) {
            log.info("未连接服务器，发送失败");
            return false;
        }

        ChannelFuture f = channel.writeAndFlush(message);
        return f.isSuccess();
    }

    public void showBanner() {
        log.info("================================================================================");
        log.info("==     ~~~   jef TCP netty client is starting    ~~~                         ==");
        log.info("================================================================================");
    }

    @Override
    public void addListener(Listener listener) {

    }

    @Override
    public void removeListener(Listener listener) {

    }
}

以上是列出了一些核心类，源码地址如下：
https://gitee.com/chengzhi2/jef-chat