实战 NettyProtostuff实现单机压测秒级接收35万个对象

2020-09-10 21:36 java进阶架构师

★★★建议星标我们★★★

公众号改版后文章乱序推荐,希望你可以点击上方“Java进阶架构师”,点击右上角,将我们设为星标”!这样才不会错过每日进阶架构文章呀。



2020年Java原创面试题库连载中

【000期】Java最全面试题库思维导图

【001期】JavaSE面试题(一):面向对象

【002期】JavaSE面试题(二):基本数据类型与访问修饰符

【003期】JavaSE面试题(三):JavaSE语法(1)

【004期】JavaSE面试题(四):JavaSE语法(3)

【005期】JavaSE面试题(五):String类

【006期】JavaSE面试题(六):泛型

【007期】JavaSE面试题(七):异常

【008期】JavaSE面试题(八):集合之List

【009期】JavaSE面试题(九):集合之Set

【010期】JavaSE面试题(十):集合之Map

【011期】JavaSE面试题(十一):多线程(1)

【012期】JavaSE面试题(十二):多线程(2)

【013期】JavaSE面试题(十三):多线程(3)

【014期】JavaSE面试题(十四):基本IO流

【015期】JavaSE面试题(十五):网络IO流

【016期】JavaSE面试题(十六):反射

【017期】JavaSE面试题(十七):JVM之内存模型

【018期】JavaSE面试题(十八):JVM之垃圾回收

【020期】JavaSE系列面试题汇总(共18篇)

【019期】JavaWeb面试题(一):JDBC

【021期】JavaWeb面试题(二):HTTP协议

【022期】JavaWeb面试题(三):Cookie和Session

【023期】JavaWeb面试题(四):JSP

【024期】JavaWeb面试题(五):Filter和Listener

【025期】Java工具面试题(一):版本控制工具

【026期】Java工具面试题(二):项目管理工具

【027期】Java设计模式面试题

【028期】JavaWeb系列面试题汇总(共10篇)

【029期】JavaEE面试题(一)Web应用服务器

【030期】JavaEE面试题(二)SpringMVC

【031期】JavaEE面试题(三)Spring(1)

【032期】JavaEE面试题(四)Spring(2)

【033期】JaveEE面试题(五)MyBatis

【034期】JavaEE面试题(六)Hibernate

【035期】JavaEE面试题(七)SpringBoot(1)

更多内容,点击上面蓝字查看


单纯netty结合protostuff进行rpc对象传输的demo网上有很多,大部分都是一个模子刻出来的,一开始我也是抄了一个,本地测试畅通无阻,未发生任何异常。

部署预发环境,进行压测后,问题巨多,各种报错层出不穷。当然,压测时我用的数据量大、发送请求非常密集,单机是每秒前100ms发送2万个对象,其他900ms歇息,死循环发送,共计40台机器作为客户端,同时往2台netty Server服务器发送对象,那么平均每个server每秒大概要接收40万个对象,由于后面还有业务逻辑,逻辑每秒只能处理35万实测。

对于网上的代码,进行了多次修改,反复测试,最终是达到了不报错无异常,单机秒级接收35万个对象以上,故写篇文章记录一下,文中代码会和线上逻辑保持一致。

Protostuff序列化和反序列化

这个没什么特殊的,网上找个工具类就好了。

引入pom

<protostuff.version>1.7.2</protostuff.version>
<dependency>
    <groupId>io.protostuff</groupId>
    <artifactId>protostuff-core</artifactId>
    <version>${protostuff.version}</version>
</dependency>

<dependency>
    <groupId>io.protostuff</groupId>
    <artifactId>protostuff-runtime</artifactId>
    <version>${protostuff.version}</version>
</dependency>

public class ProtostuffUtils {
    /**
     * 避免每次序列化都重新申请Buffer空间
     * 这句话在实际生产上没有意义,耗时减少的极小,但高并发下,如果还用这个buffer,会报异常说buffer还没清空,就又被使用了
     */

//    private static LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
    /**
     * 缓存Schema
     */

    private static Map<Class<?>, Schema<?>> schemaCache = new ConcurrentHashMap<>();
 
    /**
     * 序列化方法,把指定对象序列化成字节数组
     *
     * @param obj
     * @param <T>
     * @return
     */

    @SuppressWarnings("unchecked")
    public static <T> byte[] serialize(T obj) {
        Class<T> clazz = (Class<T>) obj.getClass();
        Schema<T> schema = getSchema(clazz);
        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
        byte[] data;
        try {
            data = ProtobufIOUtil.toByteArray(obj, schema, buffer);
//            data = ProtostuffIOUtil.toByteArray(obj, schema, buffer);
        } finally {
            buffer.clear();
        }
 
        return data;
    }
 
    /**
     * 反序列化方法,将字节数组反序列化成指定Class类型
     *
     * @param data
     * @param clazz
     * @param <T>
     * @return
     */

    public static <T> deserialize(byte[] data, Class<T> clazz) {
        Schema<T> schema = getSchema(clazz);
        T obj = schema.newMessage();
        ProtobufIOUtil.mergeFrom(data, obj, schema);
//        ProtostuffIOUtil.mergeFrom(data, obj, schema);
        return obj;
    }
 
    @SuppressWarnings("unchecked")
    private static <T> Schema<T> getSchema(Class<T> clazz) {
        Schema<T> schema = (Schema<T>) schemaCache.get(clazz);
        if (Objects.isNull(schema)) {
            //这个schema通过RuntimeSchema进行懒创建并缓存
            //所以可以一直调用RuntimeSchema.getSchema(),这个方法是线程安全的
            schema = RuntimeSchema.getSchema(clazz);
            if (Objects.nonNull(schema)) {
                schemaCache.put(clazz, schema);
            }
        }
 
        return schema;
    }
}

此处有坑,就是最上面大部分网上代码都是用了static的buffer。在单线程情况下没有问题。在多线程情况下,非常容易出现buffer一次使用后尚未被clear,就再次被另一个线程使用,会抛异常。而所谓的避免每次都申请buffer空间,实测性能影响极其微小。

另里面两次ProtostuffIOUtil都改成了ProtobufIOUtil,因为也是出过异常,修改后未见有异常。

自定义序列化方式

解码器decoder:

import com.jd.platform.hotkey.common.model.HotKeyMsg;
import com.jd.platform.hotkey.common.tool.ProtostuffUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
 
import java.util.List;
 
/**
 * @author wuweifeng
 * @version 1.0
 * @date 2020-07-29
 */

public class MsgDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf in, List<Object> list) {
        try {
 
            byte[] body = new byte[in.readableBytes()];  //传输正常
            in.readBytes(body);
 
            list.add(ProtostuffUtils.deserialize(body, HotKeyMsg.class));
 
//            if (in.readableBytes() < 4) {
//                return;
//            }
//            in.markReaderIndex();
//            int dataLength = in.readInt();
//            if (dataLength < 0) {
//                channelHandlerContext.close();
//            }
//            if (in.readableBytes() < dataLength) {
//                in.resetReaderIndex();
//                return;
//            }
//
//            byte[] data = new byte[dataLength];
//            in.readBytes(data);
//
//            Object obj = ProtostuffUtils.deserialize(data, HotKeyMsg.class);
//            list.add(obj);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

编码器 encoder


import com.jd.platform.hotkey.common.model.HotKeyMsg;
import com.jd.platform.hotkey.common.tool.Constant;
import com.jd.platform.hotkey.common.tool.ProtostuffUtils;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
 
/**
 * @author wuweifeng
 * @version 1.0
 * @date 2020-07-30
 */

public class MsgEncoder extends MessageToByteEncoder {
 
    @Override
    public void encode(ChannelHandlerContext ctx, Object in, ByteBuf out) {
        if (in instanceof HotKeyMsg) {
            byte[] bytes = ProtostuffUtils.serialize(in);
            byte[] delimiter = Constant.DELIMITER.getBytes();
 
            byte[] total = new byte[bytes.length + delimiter.length];
            System.arraycopy(bytes, 0, total, 0, bytes.length);
            System.arraycopy(delimiter, 0, total, bytes.length, delimiter.length);
 
            out.writeBytes(total);
        }
    }
}

先看Decoder解码器,这个是用来netty收到消息后,进行解码,将字节转为对象(自定义的HotKeyMsg)用的。里面有一堆被我注释掉了,注释掉的,应该在网上找到的帖子都是那么写的。这种方式本身在普通场景下是没问题的,解码还算正常,但是当上几十万时非常容易出现粘包问题。所以我是在这个解码器前增加了一个DelimiterBasedFrameDecoder分隔符解码器。

当收到消息时,先过这个分隔符解码器,之后到MsgDecoder那里时,就是已经分隔好的一个对象字节流了,就可以直接用proto工具类进行反序列化的。Constant.DELIMITER是我自定义的一个特殊字符串,用来做分隔符。

再看encoder,编码器,首先将要传输的对象用ProtostuffUtils序列化为byte[],然后在尾巴上挂上我自定义的那个分隔符。这样在对外发送对象时,就会走这个编码器,并被加上分隔符。

对应的server端代码大概是这样:

之后在Handler里就可以直接使用这个传输的对象了。

再看client端

和Server端是一样的,也是这几个编解码器,没有区别。因为netty和server之间通讯,我都是用的同一个对象定义。

同理handler也是一样的。

单机和集群

以上都写完后,其实就可以测试了,我们可以启动一个client,一个server,然后搞个死循环往Server发这个对象了,然后你在server端在收到这个对象后,再直接把这个对象也写回来,原样发送到客户端。会发现运行的很顺畅,每秒发N万个没问题,编解码都正常,client和server端都比较正常,当前前提是ProtoBuf的工具类和我的一样,不要共享那个buffer。网上找的文章基本上到这样也就结束了,随便发几个消息没问题也就算OK。然而实际上,这种代码上线后,会坑的不要不要的。

其实本地测试也很容易,再启动几个客户端,都连同一个Server,然后给他死循环发对象,再看看两端会不会有异常。这种情况下,和第一种的区别其实客户端没什么变化,Server端就有变化了,之前同时只给一个client发消息,现在同时给两个client发消息,这一步如果不谨慎就会出问题了,建议自行尝试。

之后,我们再加点料,我启动两个Server,分别用两个端口,线上其实是两台不同的server服务器,client会同时往两台server死循环发对象,如下图代码。

发消息,我们常用的就是channel.writeAndFlush(),大家可以把那个sync去掉,然后跑一下代码看看。会发现异常抛的一坨一坨的。我们明明是往两个不同的channel发消息,只不过时间是同时,结果就是发生了严重的粘包。server端收到的消息很多都是不规范的,会大量报错。如果在两个channel发送间隔100ms,情况就解决了。当然,最终我们可以使用sync同步发送,这样就不会抛异常了。

以上代码经测试,40台client,2台Server,平均每个server每秒大概接收40万个对象,可以持续稳定运行。

来源 | https://urlify.cn/Bvueuq

之前,给大家发过三份Java面试宝典,这次新增了一份,目前总共是四份面试宝典,相信在跳槽前一个月按照面试宝典准备准备,基本没大问题。

  • 《java面试宝典5.0》(初中级)

  • 《350道Java面试题:整理自100+公司》(中高级)

  • 《资深java面试宝典-视频版》(资深)

  • 《Java[BAT]面试必备》(资深)

分别适用于初中级,中高级资深级工程师的面试复习。

内容包含java基础、javaweb、mysql性能优化、JVM、锁、百万并发、消息队列,高性能缓存、反射、Spring全家桶原理、微服务、Zookeeper、数据结构、限流熔断降级等等。

获取方式:点“在看”,V信关注上述Java最全面试题库号并回复 【面试】即可领取,更多精彩陆续奉上。

看到这里,证明有所收获

必须点个在看支持呀,喵

本文章转载自公众号:java_jiagoushi

首页 - java 相关的更多文章: