SpringBoot 整合Netty自定义消息协议

本文主要介绍springboot项目，配置netty进行消息通信，自定义的netty消息协议，本文消息主要以表格中进行

消息头	消息体长度	加密标识（可选）	加密类型（可选）	消息体标识	消息体	校验码
2字节	2字节	1字节（本文不包含）	1字节（本文不包含）	2字节	N字节	2字节（本文不包含）

1. 创建普通SpringBoot项目

1.1 引入pom.xml的依赖

<dependency>
   <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
    <version>2.3.2.RELEASE</version>
</dependency>
<dependency>
    <groupId>io.netty</groupId>
    <artifactId>netty-all</artifactId>
    <version>4.1.6.Final</version>
</dependency>
<dependency>
    <groupId>com.google.code.gson</groupId>
    <artifactId>gson</artifactId>
    <version>2.8.5</version>
</dependency>
<dependency>
    <groupId>org.projectlombok</groupId>
    <artifactId>lombok</artifactId>
    <version>1.16.22</version>
</dependency>

1.2 增加application.yml文件

server:
  port: 8070
spring:
  application:
    name: netty-server
netty:
  port: 8090

1.3 创建工具类

public class GsonUtil {
    private static Gson gson = new GsonBuilder().disableHtmlEscaping().create();
    public static String toJson(Object obj) {
        return gson.toJson(obj);
    }
    public static <T> T fromJsonThrow(String str, Type type) {
        try {
            return gson.fromJson(str, type);
        }catch (Exception e){
            throw new RuntimeException(e.getMessage());
        }
    }
    public static <T> T fromJson(String str, Type type) {
        try {
            return gson.fromJson(str, type);
        }catch (Exception e){
            return null;
        }
    }
}

2. Netty 启动与消息处理

2.1 利用springboot项目启动时，启动netty

@Component
public class NettyInitializer implements InitializingBean {

    @Value("${netty.port:8090}")
    public int port;

    @Override
    public void afterPropertiesSet() throws Exception {
        start();
    }
    public void start(){
        //开启新的线程,启动netty
        new Thread(()->{
            NioEventLoopGroup bossgroup = new NioEventLoopGroup();
            NioEventLoopGroup workgroup = new NioEventLoopGroup();
            ServerBootstrap bootstrap = new ServerBootstrap();
            try {
                bootstrap.group(bossgroup,workgroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG,1024)
                        .childOption(ChannelOption.SO_KEEPALIVE,true)
                        .childHandler(new ChanneInitHandler());
                ChannelFuture channelFuture = bootstrap.bind(port).sync();
                channelFuture.channel().closeFuture().sync();
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                bossgroup.shutdownGracefully();
                workgroup.shutdownGracefully();
            }
        }).start();
    }
}

2.2 创建消息链接缓存类
请注意这样写为了快，在实际情况中，可以按需修改，与设计

public class Transmission {
    private static Map<String, Channel> channelMap = new ConcurrentHashMap<>();
    private static Map<Integer, String> classMap = new ConcurrentHashMap<Integer, String>(){{
        put(LoginMessage.CODE,LoginMessage.class.getName());
        put(LoginRespMessage.CODE,LoginRespMessage.class.getName());
        put(SmsMessage.CODE,SmsMessage.class.getName());
    }};
    public static String getClassMap(Integer str){
        return classMap.get(str);
    }
    public static Channel getChannelMap(String str){
        return channelMap.get(str);
    }
    public static Map<String, Channel> getChannelMap(){
        return channelMap;
    }
    public static void setChannelMap(String str,Channel channel){
        channelMap.put(str,channel);
    }
    public static void removeChannelMap(String str){
        channelMap.remove(str);
    }

    public static void transtion(String name,AbstractMessage message){
        ByteBuf buffer = Unpooled.buffer();
        message.encode0(buffer);
        Channel channel = channelMap.get(name);
        channel.writeAndFlush(buffer);
    }

}

2.3 创建自定义协议的消息

/**
*创建消息抽象父类
**/
@Getter
@Setter
public abstract class AbstractMessage {
    public static int HAND = 0xA55A;//消息头
    private static int LEN = 0;//长度
    //解码
    public void decode0(ByteBuf buf){
        decode(buf);
    }
    //编码
    public void encode0(ByteBuf buf){
        buf.writeShort(HAND);
        buf.writeShort(LEN);
        buf.writeShort(code());
        encode(buf);
        int newLen = buf.readableBytes();
        buf.setShort(2,newLen);
    }
    abstract int code();//消息体标识
    abstract void decode(ByteBuf buf);//抽象解码
    abstract void encode(ByteBuf buf);//抽象编码

    public String decodeStr(ByteBuf buf,int len){
        StringBuilder builder = new StringBuilder();
        for (int i = 0; i < len; i++) {
            builder.append((char) buf.readByte());
        }
        return builder.toString();
    }
    public void encodeStr(ByteBuf buf,String str,int len){
        byte[] bytes = str.getBytes();
        for (int i = 0; i < len; i++) {
            buf.writeByte(bytes[i]);
        }
    }
    @Override
    public String toString() {
        return GsonUtil.toJson(this);
    }
}

@Setter
@Getter
public class SmsMessage extends AbstractMessage{
    public static int CODE = 0x00;
    private String msaage;
    @Override
    int code() {
        return CODE;
    }
    @Override
    void decode(ByteBuf buf) {
        msaage = buf.toString(CharsetUtil.UTF_8);
    }
    @Override
    void encode(ByteBuf buf) {
        buf.writeBytes(Unpooled.copiedBuffer(msaage,CharsetUtil.UTF_8));
    }
}

@Setter
@Getter
public class LoginMessage extends AbstractMessage {
    public static int CODE = 0x01;
    private int time;//登录时间戳
    private String name;//登录人姓名
    private String userId;//登录用户id
    private int type;//操作 1,2,3,4,5,6

    @Override
    int code() {return CODE;}
    @Override
    void decode(ByteBuf buf) {
        time = buf.readInt();
        name = decodeStr(buf,10);
        userId = decodeStr(buf,10);
        type = buf.readByte();
    }

    @Override
    void encode(ByteBuf buf) {
        buf.writeInt(time);
        encodeStr(buf,name,10);
        encodeStr(buf,userId,10);
        buf.writeByte(type);
    }
}

@Setter
@Getter
public class LoginRespMessage extends AbstractMessage {
    public static int CODE = 0x02;
    private int result;//操作结果 1,2,3
    private String userId;//登录用户id
    @Override
    int code() {return CODE;}
    @Override
    void decode(ByteBuf buf) {
        result = buf.readByte();
        userId = decodeStr(buf,10);
    }
    @Override
    void encode(ByteBuf buf) {
        buf.writeByte(result);
        encodeStr(buf,userId,10);
    }
}

2.3 创建处理消息的 handler

@Slf4j
public class ReadHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        log.info("read from "+ctx.channel().remoteAddress()+" data : "+ ByteBufUtil.hexDump(msg));
        msg.retain();
        ctx.fireChannelRead(msg);
    }
}

@Slf4j
public class WriteHandler extends ChannelOutboundHandlerAdapter {
    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        if(msg instanceof ByteBuf){
            ByteBuf buf = (ByteBuf) msg;
            log.info("write to "+ctx.channel().remoteAddress()+"  data : "+ ByteBufUtil.hexDump(buf));
        }
        super.write(ctx, msg, promise);
    }
}

@Slf4j
public class LoginMessageHandler extends SimpleChannelInboundHandler<LoginMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, LoginMessage msg) throws Exception {
      log.info("LoginMessage : "+msg);
    }
}

@Slf4j
public class SmsMessageHandler extends SimpleChannelInboundHandler<SmsMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, SmsMessage msg) throws Exception {
        log.info(ctx.name()+" " + msg);
    }
}

//请注意 这里的 Transmission 类是用来简单展示链接的channel的，为了方便放在这里
@Slf4j
public class DecodeHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        Transmission.setChannelMap(ctx.name(),ctx.channel());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        AbstractMessage message = doDecode(msg);
        if(message == null){
            return;
        }
        ctx.fireChannelRead(message);
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        Transmission.removeChannelMap(ctx.name());
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }
    //消息解码为对应的消息实体
    public AbstractMessage doDecode(ByteBuf buf){
        ByteBuf buffer = Unpooled.buffer();
        if (buf.readUnsignedShort() == AbstractMessage.HAND) {
            short len = buf.readShort();
            buf.readBytes(buffer,len-4);
            int code = buffer.readUnsignedShort();
            String classMap = Transmission.getClassMap(code);
            if(classMap == null){
                log.warn("class not found");
                return null;
            }
            try {
                Class<?> clazz = Class.forName(classMap);
                AbstractMessage message = (AbstractMessage) clazz.newInstance();
                message.decode0(buffer);
                return message;
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        buffer.clear();
        return null;
    }
}

2.4 创建 ChannelInitializer 继承类

public class ChanneInitHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new IdleStateHandler(30,30,30, TimeUnit.SECONDS));
        pipeline.addLast(new ReadHandler());
        pipeline.addLast(new WriteHandler());
        pipeline.addLast(new DecodeHandler());
        pipeline.addLast(new LoginMessageHandler());
        pipeline.addLast(new SmsMessageHandler());
    }
}

3. 创建 web接口

@RestController
public class MonitorController {
    @GetMapping("/getmsg")
    public Object getmsg(){
        Map<String, Channel> channelMap = Transmission.getChannelMap();
        return channelMap;
    }
    @GetMapping("/sendmsg")
    public Object sendmsg(String msg){
        Map<String, Channel> channelMap = Transmission.getChannelMap();
        if (channelMap != null && channelMap.size() > 0) {
            for (Map.Entry<String, Channel> entry : channelMap.entrySet()) {
                SmsMessage message = new SmsMessage();
                message.setMsaage(msg);
                Transmission.transtion(entry.getKey(),message);
            }
        }
        return channelMap;
    }
}

4. 创建客户端去连接

4.1 创建客户端连接

public class Client {
    public static void main(String[] args) throws Exception {
        NioEventLoopGroup loopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(loopGroup)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE,true)
                .option(ChannelOption.SO_BACKLOG,1024)
                .handler(new ChannelInitClinet());
        bootstrap.connect("localhost",8090).sync();
    }
}

4.2 创建handler消息处理器

@Slf4j
public class DecodeHandler extends SimpleChannelInboundHandler<ByteBuf> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
        AbstractMessage message = doDecode(msg);
        if(message == null){
            return;
        }
        ctx.fireChannelRead(message);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
    }

    public AbstractMessage doDecode(ByteBuf buf){
        ByteBuf buffer = Unpooled.buffer();
        if (buf.readUnsignedShort() == AbstractMessage.HAND) {
            short len = buf.readShort();
            buf.readBytes(buffer,len-4);
            int code = buffer.readUnsignedShort();
            String classMap = Transmission.getClassMap(code);
            if(classMap == null){
                log.warn("class not found");
                return null;
            }
            try {
                Class<?> clazz = Class.forName(classMap);
                AbstractMessage message = (AbstractMessage) clazz.newInstance();
                message.decode0(buffer);
                return message;
            } catch (Exception e) {
                throw new RuntimeException(e);
            }
        }
        buffer.clear();
        return null;
    }
}

@Slf4j
public class SmsClientHandler extends SimpleChannelInboundHandler<SmsMessage> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, SmsMessage msg) throws Exception {
        log.info(ctx.name()+" " + msg);
    }
 }

4.3 创建ChannelInitializer 消息处理器

public class ChannelInitClinet extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipeline = ch.pipeline();
        pipeline.addLast(new IdleStateHandler(30,30,30, TimeUnit.SECONDS));
        pipeline.addLast(new ReadHandler());
        pipeline.addLast(new WriteHandler());
        pipeline.addLast(new DecodeHandler());
        pipeline.addLast(new SmsClientHandler());
    }
}

5. 测试连接与消息发送

5.1 启动服务端与客户端
5.1.1 在服务端调用接口，发现已经有一个连接了
在这里插入图片描述
5.1.2 代用服务端发送消息

5.1.3 客户端以接受到消息

5.1.4 停掉客户端，查看服务端缓存的channl

到这里算是ok了
以上是对自定义消息的简介，大家可以参考