Netty系列(一):Springboot整合Netty,自定义协议实现

简介: Netty是由JBOSS提供的一个java开源框架,现为 [Github](https://github.com/netty/netty)上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序。

Netty是由JBOSS提供的一个java开源框架,现为 Github上的独立项目。Netty提供异步的、事件驱动的网络应用程序框架和工具,用以快速开发高性能、高可靠性的网络服务器和客户端程序。

也就是说,Netty 是一个基于NIO的客户、服务器端的编程框架,使用Netty 可以确保你快速和简单的开发出一个网络应用,例如实现了某种协议的客户、服务端应用。Netty相当于简化和流线化了网络应用的编程开发过程。

Springboot整合Netty

新建springboot项目,并在项目以来中导入netty包,用fastjson包处理jsonStr。

        <!-- netty -->
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.42.Final</version>
        </dependency>

        <!-- Json处理 -->
        <dependency>
            <groupId>com.alibaba.fastjson2</groupId>
            <artifactId>fastjson2</artifactId>
            <version>2.0.16</version>
        </dependency>

创建netty相关配置信息文件

  1. yml配置文件——application.yml
# netty 配置
netty:
  # boss线程数量
  boss: 4
  # worker线程数量
  worker: 2
  # 连接超时时间
  timeout: 6000
  # 服务器主端口
  port: 18023
  # 服务器备用端口
  portSalve: 18026
  # 服务器地址
  host: 127.0.0.1
  1. netty配置实体类——NettyProperties与yml配置文件绑定

通过@ConfigurationProperties(prefix = "netty")注解读取配置文件中的netty配置,通过反射注入值,需要在实体类中提供对应的setter和getter方法。

@ConfigurationProperties(prefix = "netty")对应的实体类属性名称不要求一定相同,只需保证“set”字符串拼接配置文件的属性和setter方法名相同即可。
@Configuration
@ConfigurationProperties(prefix = "netty")
public class NettyProperties {

    /**
     * boss线程数量
     */
    private Integer boss;

    /**
     * worker线程数量
     */
    private Integer worker;

    /**
     * 连接超时时间
     */
    private Integer timeout = 30000;

    /**
     * 服务器主端口
     */
    private Integer port = 18023;

    /**
     * 服务器备用端口
     */
    private Integer portSalve = 18026;

    /**
     * 服务器地址 默认为本地
     */
    private String host = "127.0.0.1";
    
    // setter、getter 。。。。
}
  1. 对netty进行配置,绑定netty相关配置设置

Netty通常由一个Bootstrap开始,主要作用是配置整个Netty程序,串联各个组件,Netty中Bootstrap类是客户端程序的启动引导类,ServerBootstrap是服务端启动引导类。

@Configuration
@EnableConfigurationProperties
public class NettyConfig {
    final NettyProperties nettyProperties;

    public NettyConfig(NettyProperties nettyProperties) {
        this.nettyProperties = nettyProperties;
    }

    /**
     * boss线程池-进行客户端连接
     *
     * @return
     */
    @Bean
    public NioEventLoopGroup boosGroup() {
        return new NioEventLoopGroup(nettyProperties.getBoss());
    }

    /**
     * worker线程池-进行业务处理
     *
     * @return
     */
    @Bean
    public NioEventLoopGroup workerGroup() {
        return new NioEventLoopGroup(nettyProperties.getWorker());
    }

    /**
     * 服务端启动器,监听客户端连接
     *
     * @return
     */
    @Bean
    public ServerBootstrap serverBootstrap() {
        ServerBootstrap serverBootstrap = new ServerBootstrap()
                // 指定使用的线程组
                .group(boosGroup(), workerGroup())
                // 指定使用的通道
                .channel(NioServerSocketChannel.class)
                // 指定连接超时时间
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, nettyProperties.getTimeout())
                // 指定worker处理器
                .childHandler(new NettyServerHandler());
        return serverBootstrap;
    }
}
  1. worker处理器,初始化通道以及配置对应管道的处理器

自定义了##@##分割符,通过DelimiterBasedFrameDecoder来处理拆包沾包问题;
通过MessageDecodeHandler将接收消息解码处理成对象实例;
通过MessageEncodeHandler将发送消息增加分割符后并编码;
最后通过ServerListenerHandler根据消息类型对应处理不同消息。

public class NettyServerHandler extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        // 数据分割符
        String delimiterStr = "##@##";
        ByteBuf delimiter = Unpooled.copiedBuffer(delimiterStr.getBytes());
        ChannelPipeline pipeline = socketChannel.pipeline();
        // 使用自定义处理拆包/沾包,并且每次查找的最大长度为1024字节
        pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
        // 将上一步解码后的数据转码为Message实例
        pipeline.addLast(new MessageDecodeHandler());
        // 对发送客户端的数据进行编码,并添加数据分隔符
        pipeline.addLast(new MessageEncodeHandler(delimiterStr));
        // 对数据进行最终处理
        pipeline.addLast(new ServerListenerHandler());
    }
}
  1. 数据解码

数据解码和编码都采用UTF8格式

public class MessageDecodeHandler extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> list) throws Exception {
        ByteBuf frame = in.retainedDuplicate();
        final String content = frame.toString(CharsetUtil.UTF_8);
        Message message = new Message(content);
        list.add(message);
        in.skipBytes(in.readableBytes());
    }
}
  1. 数据解码转换的实例

Message类用于承载消息、转JsonString


public class Message {
    /**
     * 数据长度
     */
    private Integer len;

    /**
     * 接收的通讯数据body
     */
    private String content;

    /**
     * 消息类型
     */
    private Integer msgType;

    public Message(Object object) {
        String str = object.toString();
        JSONObject jsonObject = JSONObject.parseObject(str);
        msgType = Integer.valueOf(jsonObject.getString("msg_type"));
        content = jsonObject.getString("body");
        len = str.length();
    }

    public String toJsonString() {
        return "{" +
                "\"msg_type\": " + msgType + ",\n" +
                "\"body\": " + content +
                "}";
    }
    // setter、getter 。。。。
}
  1. 数据编码

netty服务端回复消息时,对消息转JsonString增加分割符,并进行编码。

public class MessageEncodeHandler extends MessageToByteEncoder<Message> {
    // 数据分割符
    String delimiter;

    public MessageEncodeHandler(String delimiter) {
        this.delimiter = delimiter;
    }

    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, Message message, ByteBuf out) throws Exception {
        out.writeBytes((message.toJsonString() + delimiter).getBytes(CharsetUtil.UTF_8));
    }
}
  1. 数据处理器,针对不同类型数据分类处理

在处理不同接收数据时使用了枚举类型,在使用switch时可以做下处理,具体参考代码,这里只演示如何操作,并没实现数据处理业务类。


public class ServerListenerHandler extends SimpleChannelInboundHandler<Message> {
    private static final Logger log = LoggerFactory.getLogger(ServerListenerHandler.class);

    /**
     * 设备接入连接时处理
     *
     * @param ctx
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) {
        log.info("有新的连接:[{}]", ctx.channel().id().asLongText());
    }

    /**
     * 数据处理
     *
     * @param ctx
     * @param msg
     */
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
        // 获取消息实例中的消息体
        String content = msg.getContent();
        // 对不同消息类型进行处理
        MessageEnum type = MessageEnum.getStructureEnum(msg);
        switch (type) {
            case CONNECT:
                // TODO 心跳消息处理
            case STATE:
                // TODO 设备状态
            default:
                System.out.println(type.content + "消息内容" + content);
        }
    }

    /**
     * 设备下线处理
     *
     * @param ctx
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) {
        log.info("设备下线了:{}", ctx.channel().id().asLongText());
    }

    /**
     * 设备连接异常处理
     *
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        // 打印异常
        log.info("异常:{}", cause.getMessage());
        // 关闭连接
        ctx.close();
    }
}
  1. 数据类型枚举类

public enum MessageEnum {
    CONNECT(1, "心跳消息"),
    STATE(2, "设备状态");

    public final Integer type;
    public final String content;

    MessageEnum(Integer type, String content) {
        this.type = type;
        this.content = content;
    }

    // case中判断使用
    public static MessageEnum getStructureEnum(Message msg) {
        Integer type = Optional.ofNullable(msg)
                .map(Message::getMsgType)
                .orElse(0);
        if (type == 0) {
            return null;
        } else {
            List<MessageEnum> objectEnums = Arrays.stream(MessageEnum.values())
                    .filter((item) -> item.getType() == type)
                    .distinct()
                    .collect(Collectors.toList());
            if (objectEnums.size() > 0) {
                return objectEnums.get(0);
            }
            return null;
        }
    }
    // setter、getter。。。。
}

到此Netty整个配置已经完成,但如果要跟随springboot一起启动,仍需要做一些配置。

  1. netty启动类配置
@Component
public class NettyServerBoot {
    private static final Logger log = LoggerFactory.getLogger(NettyServerBoot.class);
    @Resource
    NioEventLoopGroup boosGroup;
    @Resource
    NioEventLoopGroup workerGroup;
    final ServerBootstrap serverBootstrap;
    final NettyProperties nettyProperties;

    public NettyServerBoot(ServerBootstrap serverBootstrap, NettyProperties nettyProperties) {
        this.serverBootstrap = serverBootstrap;
        this.nettyProperties = nettyProperties;
    }


    /**
     * 启动netty
     *
     * @throws InterruptedException
     */
    @PostConstruct
    public void start() throws InterruptedException {
        // 绑定端口启动
        serverBootstrap.bind(nettyProperties.getPort()).sync();
        // 备用端口
        serverBootstrap.bind(nettyProperties.getPortSalve()).sync();
        log.info("启动Netty: {},{}", nettyProperties.getPort(), nettyProperties.getPortSalve());
    }

    /**
     * 关闭netty
     */
    @PreDestroy
    public void close() {
        log.info("关闭Netty");
        boosGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }
}

增加NettyServerBoot配置后,启动application时,netty服务端会跟随一起启动。
在这里插入图片描述
同时,在springboot关闭前,会先销毁netty服务。
在这里插入图片描述

完整源码

https://github.com/BerBai/JavaExample/tree/master/netty

目录
相关文章
|
2月前
|
开发框架 前端开发 网络协议
Spring Boot结合Netty和WebSocket,实现后台向前端实时推送信息
【10月更文挑战第18天】 在现代互联网应用中,实时通信变得越来越重要。WebSocket作为一种在单个TCP连接上进行全双工通信的协议,为客户端和服务器之间的实时数据传输提供了一种高效的解决方案。Netty作为一个高性能、事件驱动的NIO框架,它基于Java NIO实现了异步和事件驱动的网络应用程序。Spring Boot是一个基于Spring框架的微服务开发框架,它提供了许多开箱即用的功能和简化配置的机制。本文将详细介绍如何使用Spring Boot集成Netty和WebSocket,实现后台向前端推送信息的功能。
357 1
|
2月前
|
NoSQL Java Redis
redis的基本命令,并用netty操作redis(不使用springboot或者spring框架)就单纯的用netty搞。
这篇文章介绍了Redis的基本命令,并展示了如何使用Netty框架直接与Redis服务器进行通信,包括设置Netty客户端、编写处理程序以及初始化Channel的完整示例代码。
58 1
redis的基本命令,并用netty操作redis(不使用springboot或者spring框架)就单纯的用netty搞。
|
4月前
|
前端开发 Java Spring
springboot 整合 netty框架, 实现 心跳检测,自动重连
springboot 整合 netty框架, 实现 心跳检测,自动重连
|
7月前
|
前端开发 JavaScript Java
Springboot+Netty+WebSocket搭建简单的消息通知
这样,你就建立了一个简单的消息通知系统,使用Spring Boot、Netty和WebSocket实现实时消息传递。你可以根据具体需求扩展和改进该系统。
166 1
|
6月前
|
消息中间件 网络协议 Java
springboot+netty+kafka实现设备信息收集(完整demo复制可用)
springboot+netty+kafka实现设备信息收集(完整demo复制可用)
96 0
|
7月前
|
网络协议 Java 物联网
Spring Boot与Netty打造TCP服务端(解决粘包问题)
Spring Boot与Netty打造TCP服务端(解决粘包问题)
1056 2
|
7月前
|
Java Spring
Springboot整合Netty,自定义协议实现
以上就是在Spring Boot中整合Netty并实现自定义协议的基本步骤。你需要根据你的自定义协议的具体需求,来实现你的编码器、解码器和处理器。
275 0
|
7月前
|
前端开发 Java 数据库连接
探索Java中最常用的框架:Spring、Spring MVC、Spring Boot、MyBatis和Netty
探索Java中最常用的框架:Spring、Spring MVC、Spring Boot、MyBatis和Netty
|
网络协议 前端开发 Java
使用Springboot整合开发Netty(一个表白的小案例)
写了很久的java并发包里面的文章,今天换一个口味。很早之前学的Netty,由于最近项目经常使用到,遇到了很多坑,因此想通过一个体系教程说一下这个高性能异步通信框架Netty,这也是netty的第一篇文章。主要是通过Springboot来整合Netty实现一个最基本的案例。
670 0
|
2月前
|
JavaScript 安全 Java
如何使用 Spring Boot 和 Ant Design Pro Vue 实现动态路由和菜单功能,快速搭建前后端分离的应用框架
本文介绍了如何使用 Spring Boot 和 Ant Design Pro Vue 实现动态路由和菜单功能,快速搭建前后端分离的应用框架。首先,确保开发环境已安装必要的工具,然后创建并配置 Spring Boot 项目,包括添加依赖和配置 Spring Security。接着,创建后端 API 和前端项目,配置动态路由和菜单。最后,运行项目并分享实践心得,包括版本兼容性、安全性、性能调优等方面。
162 1