netty+springboot+oracle+protobuf 搭建客户端服务端

netty5已经不维护了，所以用netty4搭建项目

一、创建俩个基础的springboot项目

分别为netty-client和netty-server

二、添加依赖

客户端

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.6</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.netty</groupId>
    <artifactId>netty-client</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>netty-client</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- spring aop -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.69.Final</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.22</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.tomcat</groupId>
                    <artifactId>tomcat-jdbc</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!--@ConfigurationProperties注解-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <!--热部署-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
        </dependency>
        <!-- 集成mybatis -->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.3.1</version>
        </dependency>
        <!--pagehelper -->
        <dependency>
            <groupId>com.github.pagehelper</groupId>
            <artifactId>pagehelper-spring-boot-starter</artifactId>
            <version>1.2.10</version>
        </dependency>
        <!--oracle驱动 -->
        <dependency>
            <groupId>com.oracle</groupId>
            <artifactId>ojdbc6</artifactId>
            <version>11.2.0.3</version>
        </dependency>
        <!-- HikariCP 连接池依赖，从父依赖获取额版本 -->
        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
        </dependency>
        <!-- apache commons加密解密工具类 -->
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
            <version>1.10</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>3.19.1</version>
        </dependency>
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java-util</artifactId>
            <version>3.19.0</version>
        </dependency>

    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
                <configuration>
                    <!--fork :  如果没有该项配置，devtools不会起作用，即应用不会restart -->
                    <fork>true</fork>
                </configuration>
            </plugin>
        </plugins>
    </build>

</project>

服务端

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
        <version>2.5.6</version>
        <relativePath/> <!-- lookup parent from repository -->
    </parent>
    <groupId>com.netty</groupId>
    <artifactId>netty-server</artifactId>
    <version>0.0.1-SNAPSHOT</version>
    <name>netty-server</name>
    <description>Demo project for Spring Boot</description>
    <properties>
        <java.version>1.8</java.version>
    </properties>
    <dependencies>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>
        <!-- spring aop -->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-aop</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>4.1.69.Final</version>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <version>1.16.22</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-jdbc</artifactId>
            <exclusions>
                <exclusion>
                    <groupId>org.apache.tomcat</groupId>
                    <artifactId>tomcat-jdbc</artifactId>
                </exclusion>
            </exclusions>
        </dependency>
        <!--@ConfigurationProperties注解-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-configuration-processor</artifactId>
            <optional>true</optional>
        </dependency>
        <!--热部署-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-devtools</artifactId>
            <scope>runtime</scope>
        </dependency>
        <!-- 集成mybatis -->
        <dependency>
            <groupId>org.mybatis.spring.boot</groupId>
            <artifactId>mybatis-spring-boot-starter</artifactId>
            <version>1.3.1</version>
        </dependency>
        <!--pagehelper -->
        <dependency>
            <groupId>com.github.pagehelper</groupId>
            <artifactId>pagehelper-spring-boot-starter</artifactId>
            <version>1.2.10</version>
        </dependency>
        <!--oracle驱动 -->
        <dependency>
            <groupId>com.oracle</groupId>
            <artifactId>ojdbc6</artifactId>
            <version>11.2.0.3</version>
        </dependency>
        <!-- HikariCP 连接池依赖，从父依赖获取额版本 -->
        <dependency>
            <groupId>com.zaxxer</groupId>
            <artifactId>HikariCP</artifactId>
        </dependency>
        <!-- apache commons加密解密工具类 -->
        <dependency>
            <groupId>commons-codec</groupId>
            <artifactId>commons-codec</artifactId>
            <version>1.10</version>
        </dependency>
        <dependency>
            <groupId>junit</groupId>
            <artifactId>junit</artifactId>
            <version>4.12</version>
            <scope>test</scope>
        </dependency>
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java</artifactId>
            <version>3.19.1</version>
        </dependency>
        <dependency>
            <groupId>com.google.protobuf</groupId>
            <artifactId>protobuf-java-util</artifactId>
            <version>3.19.0</version>
        </dependency>
        <!--redis-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-redis</artifactId>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <plugin>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-maven-plugin</artifactId>
            </plugin>
        </plugins>
    </build>

</project>

三、yml配置

客户端

server:
  port: 8002
  servlet:
    context-path: /
netty:
  client:
    host: 127.0.0.1
    port: 8081

async:
  executor:
    thread:
      core_pool_size: 8
      max_pool_size: 8
      queue_capacity: 99999
      name:
        prefix: async-service-
pagehelper:
 helperDialect: oracle
 reasonable: true
 supportMethodsArguments: true
 params: count=countSql
mybatis:
  config-location:
    classpath: mybatis/mybatis-config.xml
  mapper-locations: mapper/*.xml
  type-aliases-package: com.netty.data
# 打印sql
logging:
  level:
    com.netty.mapper : debug
spring:
  devtools:
    restart:
      enabled: false
# 配置数据源信息
  datasource:                                           # 数据源的相关配置
    type: com.zaxxer.hikari.HikariDataSource          # 数据源类型：HikariCP
    driver-class-name: oracle.jdbc.driver.OracleDriver       # oracle驱动
    url: jdbc:oracle:thin:@127.0.0.1:1521/orcl
    username: root
    password: 123456
    hikari:
      connection-timeout: 30000        # 等待连接池分配连接的最大时长（毫秒），超过这个时长还没可用的连接则发生SQLException， 默认:30秒
      minimum-idle: 5                  # 最小连接数
      maximum-pool-size: 20            # 最大连接数
      auto-commit: true                # 事务自动提交
      idle-timeout: 600000             # 连接超时的最大时长（毫秒），超时则被释放（retired），默认:10分钟
      pool-name: DateSourceHikariCP     # 连接池名字
      max-lifetime: 1800000             # 连接的生命时长（毫秒），超时而且没被使用则被释放（retired），默认:30分钟 1800000ms
      connection-test-query: SELECT 1 FROM DUAL  # 连接测试语句

服务端

server:
  port: 8001
  servlet:
    context-path: /
netty:
  server:
    host: 127.0.0.1
    port: 8081
async:
  executor:
    thread:
      core_pool_size: 8
      max_pool_size: 8
      queue_capacity: 99999
      name:
        prefix: async-service-
mybatis:
  config-location:
    classpath: mybatis/mybatis-config.xml
  mapper-locations: mapper/*.xml
  type-aliases-package: com.netty.data
# 打印sql
logging:
  level:
    com.netty.mapper : debug
spring:
  devtools:
    restart:
      enabled: false
  # 配置数据源信息
  datasource:                                           # 数据源的相关配置
    type: com.zaxxer.hikari.HikariDataSource          # 数据源类型：HikariCP
    driver-class-name: oracle.jdbc.driver.OracleDriver       # oracle驱动
    url: jdbc:oracle:thin:@127.0.0.1:1521/orcl
    username: root
    password: 123456
    hikari:
      connection-timeout: 30000        # 等待连接池分配连接的最大时长（毫秒），超过这个时长还没可用的连接则发生SQLException， 默认:30秒
      minimum-idle: 5                  # 最小连接数
      maximum-pool-size: 20            # 最大连接数
      auto-commit: true                # 事务自动提交
      idle-timeout: 600000             # 连接超时的最大时长（毫秒），超时则被释放（retired），默认:10分钟
      pool-name: DateSourceHikariCP     # 连接池名字
      max-lifetime: 1800000             # 连接的生命时长（毫秒），超时而且没被使用则被释放（retired），默认:30分钟 1800000ms
      connection-test-query: SELECT 1 FROM DUAL  # 连接测试语句
  redis:
    host: 127.0.0.1
    port: 6379
    #password:
    timeout: 30000
    jedis:
      pool:
        max-active: 8
        max-idle: 8
        min-idle: 0
        max-wait: -1

四、目录结构

 五、代码

客户端netty代码

NettyClient
/**
 * uifuture.com
 * Copyright (C) 2013-2018 All Rights Reserved.
 */
package com.netty.client;


import com.netty.protobuf.ProtobufDecoder;
import com.netty.protobuf.ProtobufEncoder;
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.timeout.IdleStateHandler;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.concurrent.TimeUnit;

/**
 * @author
 */
@Component
public class NettyClient {

    public void start()  throws Exception{
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(group)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .remoteAddress(new InetSocketAddress("127.0.0.1", 8081))
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch)
                                throws Exception {
                            ChannelPipeline pipeline = ch.pipeline();
                            pipeline.addLast(new IdleStateHandler(0, 4, 0, TimeUnit.SECONDS));
                            pipeline.addLast(new ProtobufEncoder());
                            pipeline.addLast(new ProtobufDecoder());
                            pipeline.addLast(new ClientChannelHandler());
                        }
                    });

            // 异步操作
            ChannelFuture future = null;
            // netty 启动时如果连接失败，会断线重连sync();
            future  = bootstrap.connect().addListener(new ConnectionListener()).sync();
            // 关闭客户端
            future.channel().closeFuture().sync();
        } finally {
            //优雅关闭
           group.shutdownGracefully().sync();
        }
    }
}
ConnectionListener
package com.netty.client;

import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.springframework.stereotype.Component;

/**
 * 负责监听启动时连接失败，重新连接功能
 * @author
 */
@Component
public class ConnectionListener implements ChannelFutureListener {
    @Override
    public void operationComplete(ChannelFuture channelFuture) throws Exception {
        if (!channelFuture.isSuccess()) {
            System.out.println("-------------服务端重新连接-----------------");
            Thread.sleep(5000);
            new NettyClient().start();
        } else {
            System.err.println("服务端链接成功...");
        }
    }
}
/**
 * uifuture.com
 * Copyright (C) 2013-2018 All Rights Reserved.
 */
package com.netty.client;

import com.netty.config.ChannelContainer;
import com.netty.protobuf.MessageProto;
import com.netty.util.SpringUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author
 */
@Component
@ChannelHandler.Sharable
public class ClientChannelHandler extends ChannelInboundHandlerAdapter {


    /**
     * 通道注册
     *
     * @param ctx
     * @throws Exception
     */
    @Override
    public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
        super.channelRegistered(ctx);
    }

    /**
     * 服务器的连接被建立后调用
     * 建立连接后该 channelActive() 方法被调用一次
     *
     * @param ctx
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) {
        System.out.println("建立连接时："+new Date());
        ChannelContainer channelContainer = SpringUtil.getBean(ChannelContainer.class);
        channelContainer.setChannel(ctx.channel());
    }

    /**
     * 关闭连接时
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //使用过程中断线重连
        System.out.println("使用过程中服务端链接不上，开始重连操作...");
        Thread.sleep(1000);
        new NettyClient().start();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {
        if (obj instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) obj;
            //如果写通道处于空闲状态,就发送心跳命令
            if (IdleState.WRITER_IDLE.equals(event.state())) {
                String socketString = ctx.channel().localAddress().toString();
                MessageProto.Message ping = MessageProto.Message.newBuilder().setId(socketString).setContent("hello").build();
                ctx.channel().writeAndFlush(ping);
            }
        }
    }

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        MessageProto.Message message = (MessageProto.Message) msg;
        String socketString = ctx.channel().remoteAddress().toString();
        System.out.println("server:"+socketString+":"+message.getContent());
        ctx.fireChannelRead(msg);
    }

    /**
     * 捕获异常时调用
     *
     * @param ctx
     * @param cause
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx,
                                Throwable cause)  throws Exception{
        //记录错误日志并关闭 channel
        super.exceptionCaught(ctx, cause);
        cause.printStackTrace();
        Channel channel = ctx.channel();
        if(channel.isActive()){
            ctx.close();
        }
    }

}
SpringUtil
package com.netty.util;

import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

@Component
public class SpringUtil implements ApplicationContextAware {


    private static ApplicationContext applicationContext;
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        if(SpringUtil.applicationContext == null) {
            SpringUtil.applicationContext = applicationContext;
        }
    }

    /**获取applicationContext*/
    public static ApplicationContext getApplicationContext() {
        return applicationContext;
    }

    /**通过name获取 Bean*/
    public static Object getBean(String name){
        return getApplicationContext().getBean(name);
    }

    /**通过class获取Bean*/
    public static <T> T getBean(Class<T> clazz){
        return getApplicationContext().getBean(clazz);
    }

    /**通过name,以及Clazz返回指定的Bean*/
    public static <T> T getBean(String name,Class<T> clazz) {
        return getApplicationContext().getBean(name, clazz);
    }
}

protobuf

ProtobufDecoder
package com.netty.protobuf;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

public class ProtobufDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 标记一下当前的readIndex的位置
        in.markReaderIndex();
        // 判断包头长度
        if (in.readableBytes() < 2) {
            // 不够包头
             return;
        }
        // 读取传送过来的消息的长度。
        int length = in.readUnsignedShort();
        // 长度如果小于0
        if (length < 0) {
            // 非法数据，关闭连接
            ctx.close();
        }
        // 读到的消息体长度如果小于传送过来的消息长度
        if (length > in.readableBytes()) {
            // 重置读取位置
            in.resetReaderIndex();
            return;
        }
        ByteBuf frame = Unpooled.buffer(length);
        in.readBytes(frame);
        try {
            byte[] inByte = frame.array();
            // 字节转成对象
            MessageProto.Message msg = MessageProto.Message.parseFrom(inByte);
            if (msg != null) {
                // 获取业务消息头
                out.add(msg);
            }
        } catch (Exception e) {
            System.out.println(ctx.channel().remoteAddress() + ",decode failed."+e.getCause());
        }
    }
}
ProtobufEncoder
package com.netty.protobuf;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

public class ProtobufEncoder extends MessageToByteEncoder<MessageProto.Message> {
    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, MessageProto.Message msg, ByteBuf out) throws Exception {
        // 将对象转换为byte
        byte[] bytes = msg.toByteArray();
        // 读取消息的长度
        int length = bytes.length;
        ByteBuf buf = Unpooled.buffer(2 + length);
        // 先将消息长度写入，也就是消息头
        buf.writeShort(length);
        // 消息体中包含我们要发送的数据
        buf.writeBytes(bytes);
        out.writeBytes(buf);
    }
}

Message.proto

syntax = "proto3";
option java_outer_classname = "MessageProto";
message Message {
  string id = 1;
  string content = 2;
}

这文件需要protobuf工具编译和idea需要安装插件

 新版idea在这里安装这俩个插件

旧的需要下载好插件配到idea中

参考安装方法 https://www.freesion.com/article/39501394899/

装好插件，添加这个，有就不用加了，这样Message.proto就可以高亮显示

springboot启动类

package com.netty;

import com.netty.client.NettyClient;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Profile;
import org.springframework.scheduling.annotation.EnableScheduling;

@SpringBootApplication
@EnableScheduling
@MapperScan({"com.netty.mapper"})
public class NettyClientApplication implements CommandLineRunner {

    public static void main(String[] args) throws Exception{
        SpringApplication.run(NettyClientApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        new NettyClient().start();
    }
}

这样就可以启动了

服务端主要这几个不同，其他都一样

NettyServer
package com.netty.server;


import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.Future;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author
 */
@Component
public class NettyServer {
    /**
     * boss事件轮询线程组
     * 处理Accept连接事件的线程，这里线程数设置为1即可，netty处理链接事件默认为单线程，过度设置反而浪费cpu资源
     */
    private EventLoopGroup boss = new NioEventLoopGroup(1);

    /**
     * worker事件轮询线程组
     * 处理hadnler的工作线程，其实也就是处理IO读写 。线程数据默认为 CPU 核心数乘以2
     */
    private EventLoopGroup worker = new NioEventLoopGroup();

    /**
     * 存储client的channel
     * key:ip，value:Channel
     */
    public static Map<String, Channel> map = new ConcurrentHashMap<String, Channel>();

    @Autowired
    private ServerChannelInitializer serverChannelInitializer;

    @Value("${netty.server.port}")
    private Integer port;

    /**与客户端建立连接后得到的通道对象*/
    private Channel channel;
    /**
     * 设置服务端端口
     * @throws Exception
     */
    public  ChannelFuture start()  {
        ServerBootstrap bootstrap = new ServerBootstrap();
        //第1步定义两个线程组，用来处理客户端通道的accept和读写事件
        bootstrap.group(boss,worker)
                //第2步绑定服务端通道
                .channel(NioServerSocketChannel.class)
                //用于构造服务端套接字ServerSocket对象，标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。
                //用来初始化服务端可连接队列
                //服务端处理客户端连接请求是按顺序处理的，所以同一时间只能处理一个客户端连接，多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小。
                .option(ChannelOption.SO_BACKLOG, 128)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                //第3步绑定handler，处理读写事件，ChannelInitializer是给通道初始化
                .childHandler(serverChannelInitializer);
        // 绑定端口，同步等待成功
        ChannelFuture channelFuture1 = bootstrap.bind(port).syncUninterruptibly();
        if (channelFuture1 != null && channelFuture1.isSuccess()) {
            //获取通道
            channel = channelFuture1.channel();
            System.out.println("Netty 服务 启动 成功, port ="+ port);
        } else {
            System.out.println("Netty 服务 启动 失败");
        }
        return channelFuture1;
    }


    /**
     * 停止Netty tcp server服务
     */
    @PreDestroy
    public void destroy() {
        System.out.println("==========Netty服务退出，释放线程资源==========");
        if (channel != null) {
            channel.close();
        }
        try {
            Future<?> future = worker.shutdownGracefully().await();
            if (!future.isSuccess()) {
                System.out.println("netty tcp workerGroup shutdown fail"+ future.cause());
            }
            Future<?> future1 = boss.shutdownGracefully().await();
            if (!future1.isSuccess()) {
                System.out.println("netty tcp bossGroup shutdown fail {}"+future1.cause());
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("Netty服务关闭成功");
    }

}
ServerChannelHandler
package com.netty.server;

import com.netty.data.Msg;
import com.netty.protobuf.MessageProto;
import com.netty.service.TaskService;
import com.netty.util.GsonUtil;
import com.netty.util.SpringUtil;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.springframework.stereotype.Component;


/**
 * @author
 */
@Component
@ChannelHandler.Sharable
public class ServerChannelHandler extends ChannelInboundHandlerAdapter {

    private String beat = "hello";
    private int count = 0;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object obj) throws Exception {
        MessageProto.Message message = (MessageProto.Message) obj;
        String socketString = ctx.channel().remoteAddress().toString();
        if(beat.equals(message.getContent())){
            System.out.println("client:"+socketString+":"+message.getContent());
            MessageProto.Message pong = MessageProto.Message.newBuilder().setContent("ok").build();
            ctx.writeAndFlush(pong);
        }else {
            count ++;
            System.out.println("服务端接收客户端"+socketString+"的数据：");
            System.out.println(count+","+message.getId()+","+message.getContent());
            String content = message.getContent();
            System.out.println(content);
            Msg msg = GsonUtil.GsonToBean(content, Msg.class);
            msg.setRemoteaddress(socketString);
            TaskService taskService = SpringUtil.getBean(TaskService.class);
            taskService.insertMsg(msg);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object obj) throws Exception {

        String socketString = ctx.channel().remoteAddress().toString();
        if (obj instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) obj;
            //如果读通道处于空闲状态，说明没有接收到心跳命令
            if (IdleState.READER_IDLE.equals(event.state())) {
                if (event.state() == IdleState.READER_IDLE) {
                    System.out.println("客户端: " + socketString + " READER_IDLE 读超时");
                } else if (event.state() == IdleState.WRITER_IDLE) {
                    System.out.println("客户端: " + socketString + " WRITER_IDLE 写超时");
                } else if (event.state() == IdleState.ALL_IDLE) {
                    System.out.println("客户端: " + socketString + " ALL_IDLE 总超时");
                }
            }
        } else {
            super.userEventTriggered(ctx, obj);
        }
    }


    /**
     * @Description 客户端连接时执行，将客户端信息保存到Map中
     * @param ctx
     **/
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("通道启动");
        super.channelActive(ctx);
        System.out.println("客户端 " + getRemoteAddress(ctx) + " 链接成功");
        //往channel map中添加channel信息
        NettyServer.map.put(getIPString(ctx), ctx.channel());
    }

    /**
     * @Description 客户端断开连接时执行，将客户端信息从Map中移除
     * @param ctx
     * @Date 2019/8/28 14:22
     * @Author wuyong
     * @return
     **/
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        //删除Channel Map中的失效Client
        System.out.println("移除客户端通道："+getIPString(ctx));
        NettyServer.map.remove(getIPString(ctx));
        ctx.close();

    }
    /**
     * 异常处理
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }


    /**
     * 获取client对象：ip+port
     *
     * @param ctx
     * @return
     */
    public String getRemoteAddress(ChannelHandlerContext ctx) {
        String socketString = "";
        socketString = ctx.channel().remoteAddress().toString();
        return socketString;
    }

    /**
     * 获取client的ip
     *
     * @param ctx
     * @return
     */
    public String getIPString(ChannelHandlerContext ctx) {
        String ipString = "";
        String socketString = ctx.channel().remoteAddress().toString();
        int colonAt = socketString.indexOf(":");
        ipString = socketString.substring(1, colonAt);
        return ipString;
    }


}
ServerChannelInitializer
package com.netty.server;


import com.netty.protobuf.ProtobufDecoder;
import com.netty.protobuf.ProtobufEncoder;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.socket.SocketChannel;

import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

@Component
public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel socketChannel) throws Exception {
        ChannelPipeline pipeline = socketChannel.pipeline();
        pipeline.addLast(new IdleStateHandler(5, 0, 0, TimeUnit.MINUTES));
        pipeline.addLast(new ProtobufDecoder());
        pipeline.addLast(new ProtobufEncoder());
        //自定义Handler
        pipeline.addLast(new ServerChannelHandler());
    }
}

启动类

package com.netty;

import com.netty.server.NettyServer;
import org.mybatis.spring.annotation.MapperScan;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
@MapperScan({"com.netty.mapper"})
public class NettyServerApplication implements CommandLineRunner {

    @Autowired
    private NettyServer nettyServer;
    public static void main(String[] args) {
        SpringApplication.run(NettyServerApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        nettyServer.start();
    }

}

线程池配置类

AsyncTaskExecutorConfig
package com.netty.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author
 */
@Configuration
@EnableAsync
public class AsyncTaskExecutorConfig {

    private static final Logger logger = LoggerFactory.getLogger(AsyncTaskExecutorConfig.class);

    @Value("${async.executor.thread.core_pool_size}")
    private int corePoolSize;
    @Value("${async.executor.thread.max_pool_size}")
    private int maxPoolSize;
    @Value("${async.executor.thread.queue_capacity}")
    private int queueCapacity;
    @Value("${async.executor.thread.name.prefix}")
    private String namePrefix;

    @Bean(name = "taskExecutor")
    public Executor taskExecutor() {
        logger.info("开启线程池=====taskExecutor====");
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        //配置核心线程数
        executor.setCorePoolSize(corePoolSize);
        //配置最大线程数
        executor.setMaxPoolSize(maxPoolSize);
        //配置队列大小
        executor.setQueueCapacity(queueCapacity);
        //配置线程池中的线程的名称前缀
        executor.setThreadNamePrefix(namePrefix);

        /**
         * 拒绝处理策略
         * CallerRunsPolicy()：交由调用方线程运行，比如 main 线程。
         * AbortPolicy()：直接抛出异常。
         * DiscardPolicy()：直接丢弃。
         * DiscardOldestPolicy()：丢弃队列中最老的任务。
         */
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        //执行初始化
        executor.initialize();
        return executor;
    }
}

其他的剩余实体类，mapper,service 自己按需求写，经过测试感觉protobuf 这种数据协议传输比自定义的协议更快

以上就是全部的测试代码，仅供参考

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/h-c-g/p/15504821.html