关于java:深入理解netty从偶现宕机看netty流量控制

一、业务背景

目前挪动端的应用场景中会用到大量的音讯推送,push音讯能够帮忙经营人员更高效地实现经营指标(比方给用户推送营销流动或者揭示APP新性能)。

对于推送零碎来说须要具备以下两个个性:

  • 音讯秒级送到用户,无延时,反对每秒百万推送,单机百万长连贯。
  • 反对告诉、文本、自定义音讯透传等展示模式。正是因为以上起因,对于零碎的开发和保护带来了挑战。下图是推送零碎的简略形容(API->推送模块->手机)。

二、问题背景

推送零碎中长连贯集群在稳定性测试、压力测试阶运行一段时间后随机会呈现一个过程挂掉的状况,概率较小(频率为一个月左右产生一次),这会影响局部客户端音讯送到的时效。

推送零碎中的长连贯节点(Broker零碎)是基于Netty开发,此节点保护了服务端和手机终端的长连贯,线上问题呈现后,增加Netty内存泄露监控参数进行问题排查,察看多天但并未排查出问题。

因为长连贯节点是Netty开发,为便于读者了解,上面简略介绍一下Netty。

三、 Netty介绍

Netty是一个高性能、异步事件驱动的NIO框架,基于Java NIO提供的API实现。它提供了对TCP、UDP和文件传输的反对,作为以后最风行的NIO框架,Netty在互联网畛域、大数据分布式计算畛域、游戏行业、通信行业等取得了宽泛的利用,HBase,Hadoop,Bees,Dubbo等开源组件也基于Netty的NIO框架构建。

四、问题剖析

4.1 猜测

最后猜测是长连接数导致的,但通过排查日志、剖析代码,发现并不是此起因造成。

长连接数:39万,如下图:

每个channel字节大小1456, 按40万长连贯计算,不致于产生内存过大景象。

4.2 查看GC日志

查看GC日志,发现过程挂掉之前频繁full GC(频率5分钟一次),但内存并未升高,狐疑堆外内存泄露。

4.3 剖析heap内存状况

ChannelOutboundBuffer对象占将近5G内存,泄露起因根本能够确定:ChannelOutboundBuffer的entry数过多导致,查看ChannelOutboundBuffer的源码能够剖析出,是ChannelOutboundBuffer中的数据。

没有写出去,导致始终积压;ChannelOutboundBuffer外部是一个链表构造。

4.4 从上图剖析数据未写出去,为什么会呈现这种状况?

代码中理论有判断连贯是否可用的状况(Channel.isActive),并且会对超时的连贯进行敞开。从历史教训来看,这种状况产生在连贯半关上(客户端异样敞开)的状况比拟多—单方不进行数据通信无问题。

按上述猜测,测试环境进行重现和测试。

1)模仿客户端集群,并与长连贯服务器建设连贯,设置客户端节点的防火墙,模仿服务器与客户端网络异样的场景(即要模仿Channel.isActive调用胜利,但数据理论发送不进来的状况)。

2)调小堆外内存,继续发送测试音讯给之前的客户端。音讯大小(1K左右)。

3)依照128M内存来计算,实际上调用9W屡次就会呈现。

五、问题解决

5.1 启用autoRead机制

当channel不可写时,敞开autoRead;

public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
    if (!ctx.channel().isWritable()) {
        Channel channel = ctx.channel();
        ChannelInfo channelInfo = ChannelManager.CHANNEL_CHANNELINFO.get(channel);
        String clientId = "";
        if (channelInfo != null) {
            clientId = channelInfo.getClientId();
        }

        LOGGER.info("channel is unwritable, turn off autoread, clientId:{}", clientId);
        channel.config().setAutoRead(false);
    }
}

当数据可写时开启autoRead;

@Override
public void channelWritabilityChanged(ChannelHandlerContext ctx) throws Exception
{
    Channel channel = ctx.channel();
    ChannelInfo channelInfo = ChannelManager.CHANNEL_CHANNELINFO.get(channel);
    String clientId = "";
    if (channelInfo != null) {
        clientId = channelInfo.getClientId();
    }
    if (channel.isWritable()) {
        LOGGER.info("channel is writable again, turn on autoread, clientId:{}", clientId);
        channel.config().setAutoRead(true);
    }
}

阐明:

autoRead的作用是更准确的速率管制,如果关上的时候Netty就会帮咱们注册读事件。当注册了读事件后,如果网络可读,则Netty就会从channel读取数据。那如果autoread关掉后,则Netty会不注册读事件。

这样即便是对端发送数据过去了也不会触发读事件,从而也不会从channel读取到数据。当recv_buffer满时,也就不会再接收数据。

5.2 设置高下水位

serverBootstrap.option(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(1024 * 1024, 8 * 1024 * 1024));

注:高下水位配合前面的isWritable应用

5.3 减少channel.isWritable()的判断

channel是否可用除了校验channel.isActive()还须要加上channel.isWrite()的判断,isActive只是保障连贯是否激活,而是否可写由isWrite来决定。

private void writeBackMessage(ChannelHandlerContext ctx, MqttMessage message) {
    Channel channel = ctx.channel();
    //减少channel.isWritable()的判断
    if (channel.isActive() && channel.isWritable()) {
        ChannelFuture cf = channel.writeAndFlush(message);
        if (cf.isDone() && cf.cause() != null) {
            LOGGER.error("channelWrite error!", cf.cause());
            ctx.close();
        }
    }
}

注:isWritable能够来管制ChannelOutboundBuffer,不让其无限度收缩。其机制就是利用设置好的channel高下水位来进行判断。

5.4 问题验证

批改后再进行测试,发送到27W次也并不报错;

六、解决思路剖析

个别Netty数据处理流程如下:将读取的数据交由业务线程解决,解决实现再发送进来(整个过程是异步的),Netty为了进步网络的吞吐量,在业务层与socket之间减少了一个ChannelOutboundBuffer。

在调用channel.write的时候,所有写出的数据其实并没有写到socket,而是先写到ChannelOutboundBuffer。当调用channel.flush的时候才真正的向socket写出。因为这两头有一个buffer,就存在速率匹配了,而且这个buffer还是无界的(链表),也就是你如果没有管制channel.write的速度,会有大量的数据在这个buffer里沉积,如果又碰到socket写不出数据的时候(isActive此时判断有效)或者写得慢的状况。

很有可能的后果就是资源耗尽,而且如果ChannelOutboundBuffer寄存的是DirectByteBuffer,这会让问题更加难排查。

流程可形象如下:

从下面的剖析能够看出,步骤一写太快(快到解决不过去)或者上游发送不出数据都会造成问题,这理论是一个速率匹配问题。

七、Netty源码阐明

超过高水位

当ChannelOutboundBuffer的容量超过高水位设定阈值后,isWritable()返回false,设置channel不可写(setUnwritable),并且触发fireChannelWritabilityChanged()。

private void incrementPendingOutboundBytes(long size, boolean invokeLater) {
    if (size == 0) {
        return;
    }

    long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, size);
    if (newWriteBufferSize > channel.config().getWriteBufferHighWaterMark()) {
        setUnwritable(invokeLater);
    }
}
private void setUnwritable(boolean invokeLater) {
    for (;;) {
        final int oldValue = unwritable;
        final int newValue = oldValue | 1;
        if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
            if (oldValue == 0 && newValue != 0) {
                fireChannelWritabilityChanged(invokeLater);
            }
            break;
        }
    }
}

低于低水位

当ChannelOutboundBuffer的容量低于低水位设定阈值后,isWritable()返回true,设置channel可写,并且触发fireChannelWritabilityChanged()。

private void decrementPendingOutboundBytes(long size, boolean invokeLater, boolean notifyWritability) {
    if (size == 0) {
        return;
    }

    long newWriteBufferSize = TOTAL_PENDING_SIZE_UPDATER.addAndGet(this, -size);
    if (notifyWritability && newWriteBufferSize < channel.config().getWriteBufferLowWaterMark()) {
        setWritable(invokeLater);
    }
}
private void setWritable(boolean invokeLater) {
    for (;;) {
        final int oldValue = unwritable;
        final int newValue = oldValue & ~1;
        if (UNWRITABLE_UPDATER.compareAndSet(this, oldValue, newValue)) {
            if (oldValue != 0 && newValue == 0) {
                fireChannelWritabilityChanged(invokeLater);
            }
            break;
        }
    }
}

八、总结

当ChannelOutboundBuffer的容量超过高水位设定阈值后,isWritable()返回false,表明音讯产生沉积,须要升高写入速度。

当ChannelOutboundBuffer的容量低于低水位设定阈值后,isWritable()返回true,表明音讯过少,须要进步写入速度。通过以上三个步骤批改后,部署线上察看半年未产生问题呈现。

​作者:vivo互联网服务器团队-Zhang Lin

【腾讯云】轻量 2核2G4M,首年65元

阿里云限时活动-云数据库 RDS MySQL  1核2G配置 1.88/月 速抢

本文由乐趣区整理发布,转载请注明出处,谢谢。

您可能还喜欢...

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据