java网络编程之Netty

基础概念

BIO：同步阻塞式IO，服务器实现模式为一个连接一个线程，即客户端有连接请求时服务器端就需要启动一个线程进行处理，如果这个连接不做任何事情会造成不必要的线程开销，当然可以通过线程池机制改善。
NIO：同步非阻塞式IO，服务器实现模式为一个请求一个线程，即客户端发送的连接请求都会注册到多路复用器上，多路复用器轮询到连接有I/O请求时才启动一个线程进行处理。
Java AIO(NIO.2) ： 异步非阻塞，服务器实现模式为一个有效请求一个线程，客户端的I/O请求都是由OS先完成了再通知服务器应用去启动线程进行处理
NIO方式适用于连接数目多且连接比较短（轻操作）的架构，比如聊天服务器，并发局限于应用中，编程比较复杂，JDK1.4开始支持。
AIO方式使用于连接数目多且连接比较长（重操作）的架构，比如相册服务器，充分调用OS参与并发操作，编程比较复杂，JDK7开始支持

网络编程注意事项

1. 黏包、断包问题解决,网络上很多DEMO都不是很完善
2. 断线重连 客户端读取数据出现异常间隔5秒重连，客户端心跳判断连接断开间隔5秒重连，发送数据失败重连（没有连上，间隔5秒重连）。
3. 业务数据包补发。考虑到网络环境的不稳定性、多变性（比如从进入电梯、进入地铁、移动网络切换到wifi等），在消息发送的时候，发送失败的概率其实不小，这时消息重发机制就很有必要了。发送消息时，除了心跳消息、握手消息、状态报告消息外，消息都加入消息发送超时管理器，立马开启一个定时器，比如每隔5秒执行一次，共执行3次，在这个周期内，如果消息没有发送成功，会进行3次重发，达到3次重发后如果还是没有发送成功，那就放弃重发，移除该消息，同时通过消息转发器通知应用层，由应用层决定是否再次重发。如果消息发送成功，服务端会返回一个消息发送状态报告，客户端收到该状态报告后，从消息发送超时管理器移除该消息，同时停止该消息对应的定时器即可。客户端发送消息，如果服务端没有反馈还可以在客户端做补发，失败的地方把该消息加入到队列，连接成功后查看队列从队列中取出补发该消息。在用户握手认证成功时，应该检查消息发送超时管理器里是否有发送超时的消息，如果有，则全部重发。还可以在发送失败报异常的时候把该消息加入到重发队列，重连成功后补发。
4. java原生的BIO开发,一般会专门启动一个线程负责读取数据，其实读写就是对socket的inputStream、outputStream进行操作。把读取到的完整包解析出来给业务线程池处理，把剩余的不完整包跟下一次读取到的数据合并。nio一般是通过类似事件通知机制实现不阻塞。
5. 心跳包：主要是为了防止NAT超时（比如手机连接互联网，运营商的网关就做了ＮＡＴ映射，把无线网跟因特网对接，隔一段时间没有通信链接会被网关释放。防火墙也会把一段时间内没通信的链接断开），探测连接是否断开，由客户端发送比较合理。链路断开, 没有写操作的TCP连接是感知不到的, 除非这个时候发送数据给服务器, 造成写超时, 否则TCP连接不会知道断开了。主动kill掉一方的进程, 另一方会关闭TCP连接, 是系统代进程给服务器发的FIN. TCP连接就是这样, 只有明确的收到对方发来的关闭连接的消息(收到RST也会关闭, 大家都懂), 或者自己意识到发生了写超时, 否则它认为连接还存在。但是网路复杂中途出现的问题也会比较常见，譬如网线被掐断，对象进程被杀掉，频繁丢包，对方这时候的TCP长连接是不可使用的，但是对于应用层并不知道。如果需要知道当前的网络状况则需要很复杂的超时进行了解，TCP底层就实现了这样的功能，心跳机制是TCP在一段时间间隔后发送确定连接是否存在，如果确定存在的话，就会回传一个包来确定连接是存在的，如果没有返回包的话，则应该通知上层，网络出现了问题，需要进行连接失败的操作了。手动关闭客户端进程，事实上并不能测试出想要的结果，因为进程是在应用层的，所以，这种测试方法不能保证网络驱动层也不发送数据报文给服务器。经过测试发现，当应用层强制结束进程时，对于TCP连接，驱动层会发送reset数据包！而服务器收到这个数据包就可以正常关闭了！那么，如果拔掉网线呢，服务器收不到这个数据包，就会导致死连接存在！所以，心跳包是必要的，或者应用TCP协议本身的Keep-alive来设置SO_KEEPALIVE。减轻服务端压力服务端可以不回复心跳包，服务端一定时间内没有收到心跳包就断开该连接。
6. 分包的方式：包头+长度、固定长度、分隔符（回车、自定义字符串）

常用方法

分包封装

• 分隔符：DelimiterBasedFrameDecoder
• 固定长度：FixedLengthFrameDecoder
• 按行分隔：LineBasedFrameDecoder 遇到一个换行符,则认为是一个完整的报文
• 自定义长度帧解码器：LengthFieldBasedFrameDecoder 可以指定最大包长度、长度域大小与位置，new LengthFieldBasedFrameDecoder(65536，0，2)
  1) lengthFieldOffset = 0；
  2) lengthFieldLength = 2；
  3) lengthAdjustment = 0；
  4) initialBytesToStrip = 0。

编解码

• StringDecoder、StringEncoder

网络编程可能问题

多进程通过数据库进行交互

通过数据库进行交互可能出现一个进程往数据库里写入了一条记录，通知另一个进程去读取。但是写入操作还未提交另一个进程就去读取数据了，读取到的数据为空，特别是在需要批量操作的场景容易出现。所以需要根据情况写入后进行提交操作，提交成功然后再通知另一个进程读取。

@Transactional(propagation=Propagation.NEVER) // java spring ibatis 提交的代码

<tx:annotation-driven transaction-manager="transactionManagerIT"/>

或者直接通过配置

<bean id="baseTransactionProxyIT" class="org.springframework.transaction.interceptor.TransactionProxyFactoryBean" abstract="true">
		<property name="transactionManager">
			<ref bean="transactionManagerIT" />
		</property>
		<property name="transactionAttributes">
			<props>
				<prop key="save*">PROPAGATION_REQUIRED,-Exception</prop>
				<prop key="send">PROPAGATION_NEVER,readOnly,-Exception</prop>
				<prop key="*">PROPAGATION_SUPPORTS,-Exception</prop>
			</props>
		</property>
	</bean>

Netty 主要功能介绍

心跳

使用空闲时间发送心跳。

public IdleStateHandler(int readerIdleTimeSeconds, int writerIdleTimeSeconds, int allIdleTimeSeconds) {
    this((long)readerIdleTimeSeconds, (long)writerIdleTimeSeconds, (long)allIdleTimeSeconds, TimeUnit.SECONDS);
}

• readerIdleTimeSeconds: 读超时. 即当在指定的时间间隔内没有从 Channel 读取到数据时, 会触发一个 READER_IDLE 的 IdleStateEvent 事件.
• writerIdleTimeSeconds: 写超时. 即当在指定的时间间隔内没有数据写入到 Channel 时, 会触发一个 WRITER_IDLE 的 IdleStateEvent 事件.
• allIdleTimeSeconds: 读/写超时. 即当在指定的时间间隔内没有读或写操作时, 会触发一个 ALL_IDLE 的 IdleStateEvent 事件.

语法说明

f.channel().closeFuture().sync();//等待服务端关闭端口监听。主线程执行到这里就 wait 子线程结束，子线程才是真正监听和接受请求的，closeFuture()是开启了一个channel的监听器，负责监听channel是否关闭的状态，如果监听到channel关闭了，子线程才会释放

与spring集成

//  ==============================存在问题=======================================================
@Slf4j
@Component
public class EchoServer {
	/**
     * NioEventLoop并不是一个纯粹的I/O线程，它除了负责I/O的读写之外
     * 创建了两个NioEventLoopGroup，
     * 它们实际是两个独立的Reactor线程池。
     * 一个用于接收客户端的TCP连接，
     * 另一个用于处理I/O相关的读写操作，或者执行系统Task、定时任务Task等。
     */
    private final EventLoopGroup bossGroup = new NioEventLoopGroup();
    private final EventLoopGroup workerGroup = new NioEventLoopGroup();
    private Channel channel;
	  public ChannelFuture  start(int port) throws Exception {
		//绑定对应ip和端口，同步等待成功。一定要用端口的形式。网络上很多用.localAddress(new InetSocketAddress(ip,port))这种形式对监听IP进行了限制，如果设置成127.0.0.1外部客户端无法访问
		ChannelFuture future = serverBootstrap.bind(port).sync();
		LOGGER.info("rpc server 已启动，端口：{}", port);
  	}
  	public void destroy() {
  	}
}

/**
* 将Netty服务端从main函数启动方式改为交给Spring来管理启动和销毁的工作。本写法只支持jar包在命令行启动，在tomcat中启动会导致tomcat假死。
*/
@SpringBootApplication
public class SpringNettyApplication implements CommandLineRunner {

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

    @Value("${netty.url}")
    private String url;

    @Autowired
    private EchoServer echoServer;

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

    @Override
    public void run(String... args) throws Exception { // CommandLineRunner run 等待Spring加载完成后执行nettyServer启动工作。tomcat中启动会有问题，新启线程执行也不行，需要异步处理，后面有例子说明
        ChannelFuture future = echoServer.start(url,port);
        Runtime.getRuntime().addShutdownHook(new Thread(){ // 这个方法的意思就是在jvm中增加一个关闭的钩子，当jvm关闭的时候，会执行系统中已经设置的所有通过方法addShutdownHook添加的钩子，当系统执行完这些钩子后，jvm才会关闭。所以这些钩子可以在jvm关闭的时候进行内存清理、对象销毁等操作
            @Override
            public void run() {
                echoServer.destroy();
            }
        });
        //服务端管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程。不能写在echoServer.start方法里面，会导致springboot启动的时候阻塞住.主线程执行到这里就 wait 子线程结束，子线程才是真正监听和接受请求的，子线程就是Netty启动的监听端口的线程； 即closeFuture()是开启了一个channel的监听器，负责监听channel是否关闭的状态，如果未来监听到channel关闭了，子线程才会释放，syncUninterruptibly()让主线程同步等待子线程结果。.channel.close()才是主动关闭通道的方法。
        future.channel().closeFuture().syncUninterruptibly();
    }
}
/**
* 在springboot+netty+tomcat中启动，需要使用线程池新建线程启动netty，可以在Listener中启动
* 遇到tomcat假死排查步骤主要是：
* 1.查看网络问题 
* 2.排查防火墙 
* 3.查看springboot jdk tomcat netty版本与依赖 
* 4.如果是在IDE中要判断是否IDE出问题 
* 5.判断程序执行到哪里假死最终定位是netty问题，服务端绑定端口后异步监听,等待客户端连接，而这个过程线程会变为wait状态
* 解决思路：新启动线程启动netty服务端监听端口
*/
/**
 * @ description 解决项目打war包发布在tomcat，端口被占用问题
 * 解决思路：给NettyServer分配一个独立的线程用于加载
 * 在IOC的容器的启动过程，当所有的bean都已经处理完成之后，spring ioc容器会有一个发布事件的动作，可以去做一些自己想做的事
 * 让我们的bean实现ApplicationListener接口，这样当发布事件时，ioc容器就会以容器的实例对象作为事件源类，并从中找到事件的监听者，此时ApplicationListener接口实例中的onApplicationEvent(E event)方法就会被调用，我们的逻辑代码就会写在此处
 * 一个是root application context ,另一个就是我们自己的 projectName-servlet context，这种情况下，就会造成onApplicationEvent方法被执行两次
 * 我们可以只在root application context初始化完成后调用逻辑代码
 * ContextStartedEvent(容器启动)、ContextRefreshedEvent(容器刷新也就是初始化完成,spring容器加载完毕做一件事情)等
 * ContextRefreshedEvent详细说明：
 * Published when the ApplicationContext is initialized or refreshed, for example, using the refresh() method on the ConfigurableApplicationContext interface. "Initialized" here means that all beans are loaded, post-processor beans are detected and activated, singletons are pre-instantiated, and the ApplicationContext object is ready for use. As long as the context has not been closed, a refresh can be triggered multiple times, provided that the chosen ApplicationContext actually supports such "hot" refreshes. For example, XmlWebApplicationContext supports hot refreshes, but GenericApplicationContext does not.
 * 当容器中的bean初始化好的时候，会回调ContextRefreshedEvent事件一次，也可以手动调用容器提供的refresh()方法，去触发ContextRefreshedEvent事件
 */
// 例子一 =====================================================================================
@Slf4j
@Component
public class EchoServer implements Runnable{
    ...
    @Override
    public void run() { // 直接把上面demo启动方法拿过来，或者自己写启动与关闭方法进行调用
        try {
            EchoServer myself = this; // 注意：不需要
            ChannelFuture future = startListener();
            Runtime.getRuntime().addShutdownHook(new Thread() { // 注意：使用Listener在tomcat下启动不需要加这个方法，无法销毁
                @Override
                public void run() {
                    myself.destroy(); // tomcat下无法进行销毁
                }
            });
            //服务端管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程
            future.channel().closeFuture().syncUninterruptibly();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    /** 正确写法
    @Override
    public void run() { 
        try {
            ChannelFuture future = startListener();
            //服务端管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程
            future.channel().closeFuture().syncUninterruptibly();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    */
}
@Component
@Log
public class NettyServerListener  implements ApplicationListener<ContextRefreshedEvent> {
    @Autowired
    private EchoServer nettyServer;
    /**
     * 当一个applicationContext被初始化或被刷新时触发
     * @param event
     */
    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext().getParent() == null) { // root application context 没有parent
            log.info("EchoServer Start Success");
            //自己的NettyServer
            // NettyServer nettyServer = new NettyServer(); // new 的方式不受spring监管、从配置文件注入的端口会无效
            new Thread(nettyServer).start();
        }
    }
    @PreDestroy
    public void cleanup() {
        nettyServer.destroy(); // tomcat可以进行销毁操作，类似于操作了一个falg标记使线程退出。但是感觉还有其他改进方案，应该终止线程，回收线程所有资源
        // 使用线程的interrupt也无法正常结束
    }
}
// ===============================可行方案======================================================
/**
* 方案二
* 可能初始化容器后调用CommandLineRunner的run方法太久导致后续无法执行,run方法如果有阻塞方法得用异步执行
* 修改方案：启动方法改成异步，也可以解决问题
*/
public class NettyServer {
    @Override
    public void start() { // 直接把上面demo启动方法拿过来，或者自己写启动与关闭方法进行调用
        try {
            ChannelFuture future = startListener();
            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    this.destroy();
                }
            });
            //服务端管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程
            future.channel().closeFuture().syncUninterruptibly();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
public class StartCommand implements CommandLineRunner {

    @Resource
    private NettyServer nettyServer;


    @Override
    public void run(String... args) throws Exception {
        CompletableFuture.runAsync(() -> nettyServer.start()); // nettyServer 不需要实现Runnable，普通类调用就行
    }

    /**
    // 通过@Async注解也能解决问题
    @Override
    @Async
    public void run(String... args) throws Exception {
        try {
            ChannelFuture future = nettyServer.startListener();
            Runtime.getRuntime().addShutdownHook(new Thread() {
                @Override
                public void run() {
                    nettyServer.destroy();
                }
            });
            //服务端管道关闭的监听器并同步阻塞,直到channel关闭,线程才会往下执行,结束进程
            future.channel().closeFuture().syncUninterruptibly();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    */
}
// =====================================================================================
// 其他例子
@WebListener
public class NettyServerListener implements //ApplicationListener<ContextStartedEvent>
   ServletContextListener // ServletContextListener用于监听tomcat启动或者关闭
{
    private static final Logger log= LoggerFactory.getLogger(NettyServerListener.class);
    private ExecutorService webSocketSinglePool;
    private  EchoServer echoServer= null;
    @Override
    public void contextInitialized(ServletContextEvent sce) {
         
    }
    @Override
    public void contextDestroyed(ServletContextEvent sce) {
        
        new Thread(()->echoServer.destroy()).run();
    }

    @PostConstruct
    public  void setup() {
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("webSocketSinglePool-%d").build();
        webSocketSinglePool = new ThreadPoolExecutor(1, 1, 0L,
                TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024),
                namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
        log.info("webSocketSinglePool init.");
    }

    /*@Override
    public void onApplicationEvent(ContextStartedEvent event) {
        log.info("监听到事件了");
        runWebSocketServer(event.getApplicationContext());
    }*/
    private void runWebSocketServer(ApplicationContext applicationContext) {
        final EchoServer echoServer = applicationContext.getBean(EchoServer.class);
       new Thread(() -> {
            try { //开始启动netty服务
                log.info("ready to start Netty");
                echoServer.run();
            } catch (Exception e) {
                log.error("webSocket listen and serve error.", e);
            }
        }).run();
    }
    @PreDestroy
    public void  cleanup(){
        webSocketSinglePool.shutdown();
        log.info("webSocketSinglePool destroyed.");
    }
}
/**
* 另一个demo
*/
@Component
public class ApplicationRefreshListener implements ApplicationListener<ContextStartedEvent> {

private static final Logger LOG = LoggerFactory.getLogger(ApplicationRefreshListener.class);
private ExecutorService webSocketSinglePool;

@PostConstruct
public void setup() {
  ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
  .setNameFormat("webSocketSinglePool-%d").build();
  webSocketSinglePool = new ThreadPoolExecutor(1, 1, 0L,
  TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024),
  namedThreadFactory, new ThreadPoolExecutor.AbortPolicy());
  LOG.info("webSocketSinglePool init.");
}

@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
  runWebSocketServer(event.getApplicationContext());
}

private void runWebSocketServer(ApplicationContext applicationContext) {
  final WebSocketServer webSocketServer = applicationContext.getBean(WebSocketServer.class);
  webSocketSinglePool.execute(() -> {
    try {
      webSocketServer.listenAndServe();
    } catch (Exception e) {
      LOG.error("webSocket listen and serve error.", e);
    }
  });
}

@PreDestroy
public void cleanup() {
  webSocketSinglePool.shutdown();
    LOG.info("webSocketSinglePool destroyed.");
  }
}
// =====================================================================================

注意要点

• 服务端使用serverBootstrap.bind(port).sync();形式监听指定端口，不要用new InetSocketAddress(ip,port)这种限制IP的形式
• Netty中ByteToMessageDecoder中的decode方法执行多次的问题,再查看decode方法的源码注释如下：This method will be called till either the input has nothing to read，意思是说：ByteBuf对象的数据没有读完的话，decode方法会一直调用。readerIndex会随着数据的读取而不断增加,所以保证每次decode读取一个完整包，如果不满足一个完整包就重置readerIndex退出下次再进来读。
• 如果使用的是分隔符的数据包还可以先查找分隔符，如果没有找到分隔符却发现包里有内容可以进行跳过操作，丢弃这些错误数据，找到分隔符了就可以进行读取，验证长度，读取包内容等操作，读取完一个完整包退出等待系统自动调用decode继续后续包操作，当然也可以自己代码判断后面如果还有内容直接进行后续包处理。
• 注意不要把没用到的netty自带的decoder解码器加上去，要根据具体需求加，不然会导致解码错误。还要注意decoder是否是可共享的如果是不可共享的得用new
• write read 数据类型默认使用ByteBuf
• 应用开发的时候往往是异步的，可能要经过多个应用，所以需要整条链路的标识，可以使用sessionid、token、id等做标识，这样返回的时候才能找到到底是谁发送的这个请求，也可以各个子应用自己生成标识单独维护各自请求的对应关系

其他扩展框架

• SOFABolt:蚂蚁金融服务集团开发的一套基于 Netty 实现的网络通信框架
• smart-socket
• t-io
• mina

参考

• Netty实现心跳机制与断线重连
• 开源一个自用的Android IM库，基于Netty+TCP+Protobuf实现
• netty 例子 example
• netty 例子 Netty-study
• netty 例子 NettyChat
• netty 例子Jupiter
• 一起学Netty（十四）之 Netty生产级的心跳和重连机制
• 浅析 Netty 实现心跳机制与断线重连
• SpringBoot使用netty服务端和客户端-这个启动写法有问题再tomcat下无法使用，new出来的不受spring管理，而且只IP监听127.0.0.1，要用CommandLineRunner
• Http-Netty-Rpc-Service系统改造
• 【Netty】Springboot整合Netty
• 【Netty】心跳机制与断线重连
• Netty与Spring Boot的整合
• Netty之解决TCP粘包拆包(自定义协议)
• Python基础-Socket基础-1
• 【Netty】Netty之ByteBuf
• 什么是编解码器？
• java netty之ByteToMessageDecoder
• Netty5 Read事件处理过程_源码讲解
• 使用LengthFieldBasedFrameDecoder解码器及自定义