Protostuff を使用した Serialize_html/css_WEB-ITnose

Sequence

rpc 呼び出しには、java のデフォルトのシリアル化 (hessian など) など、java で使用される json、bson などの多くのシリアル化メソッドがあります。 -言語のもの（thriftやprotocolbufなど）。 thrift と pb の利点はシリアル化後のサイズが比較的小さいことですが、欠点は Java コードを生成する必要があり、まったく役に立たないため、実行時に両者がどれほど効率的であっても、開発効率が比較的低いことです。 hessian と同様にいくつか使用されていますが、pb ほどの強力さは感じられません。そこで、業務効率と開発効率を両立するシリアル化方法を模索してきました。時々インターネットで protostuff を見て、探していたシリアル化方法を見つけたような気がしました。

protostuff の概要

protobuf の欠点の 1 つは、データ構造の事前コンパイル プロセスが必要であることです。まず、構成ファイルを .proto 形式で作成してから使用する必要があります。さまざまな言語応答のコードを生成するために protobuf によって提供されるツール。 Java にはリフレクションと動的コード生成の機能があるため、このプリコンパイル プロセスは必要なく、コードの実行時に実装できます。この機能をすでに実装している protostuff があります。

プロトスタッフ効率

• Ser Time+Deser Time (ns)

• サイズ、圧縮サイズ[light] バイト単位

使用

pom 依存関係

        <dependency>            <groupId>com.dyuproject.protostuff</groupId>            <artifactId>protostuff-core</artifactId>            <version>1.0.8</version>        </dependency>        <dependency>            <groupId>com.dyuproject.protostuff</groupId>            <artifactId>protostuff-runtime</artifactId>            <version>1.0.8</version>        </dependency>

ツール クラス

public class SerializationUtil {    private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<Class<?>, Schema<?>>();    private static Objenesis objenesis = new ObjenesisStd(true);    private static <T> Schema<T> getSchema(Class<T> clazz) {        @SuppressWarnings("unchecked")        Schema<T> schema = (Schema<T>) cachedSchema.get(clazz);        if (schema == null) {            schema = RuntimeSchema.getSchema(clazz);            if (schema != null) {                cachedSchema.put(clazz, schema);            }        }        return schema;    }    /**     * 序列化     *     * @param obj     * @return     */    public static <T> byte[] serializer(T obj) {        @SuppressWarnings("unchecked")        Class<T> clazz = (Class<T>) obj.getClass();        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);        try {            Schema<T> schema = getSchema(clazz);            return ProtostuffIOUtil.toByteArray(obj, schema, buffer);        } catch (Exception e) {            throw new IllegalStateException(e.getMessage(), e);        } finally {            buffer.clear();        }    }    /**     * 反序列化     *     * @param data     * @param clazz     * @return     */    public static <T> T deserializer(byte[] data, Class<T> clazz) {        try {            T obj = objenesis.newInstance(clazz);            Schema<T> schema = getSchema(clazz);            ProtostuffIOUtil.mergeFrom(data, obj, schema);            return obj;        } catch (Exception e) {            throw new IllegalStateException(e.getMessage(), e);        }    }}

netty ベースの rpc

• NettyServer

public class NettyServer {    private static final Logger logger = LoggerFactory.getLogger(NettyServer.class);    private int ioThreadNum;    //内核为此套接口排队的最大连接个数，对于给定的监听套接口，内核要维护两个队列，未链接队列和已连接队列大小总和最大值    private int backlog;    private int port;    private Channel channel;    private EventLoopGroup bossGroup;    private EventLoopGroup workerGroup;    public NettyServer(int ioThreadNum, int backlog, int port) {        this.ioThreadNum = ioThreadNum;        this.backlog = backlog;        this.port = port;    }    public void start() throws InterruptedException {        bossGroup = new NioEventLoopGroup();        workerGroup = new NioEventLoopGroup(this.ioThreadNum);        final Map<String,Object> demoService = new HashMap<String, Object>();        demoService.put("com.patterncat.service.HelloService", new HelloServiceImpl());        ServerBootstrap serverBootstrap = new ServerBootstrap();        serverBootstrap.group(bossGroup, workerGroup)                .channel(NioServerSocketChannel.class)                .option(ChannelOption.SO_BACKLOG, backlog)                //注意是childOption                .childOption(ChannelOption.SO_KEEPALIVE, true)                .childOption(ChannelOption.TCP_NODELAY, true)                .childHandler(new ChannelInitializer<SocketChannel>() {                    @Override                    protected void initChannel(SocketChannel socketChannel) throws Exception {                        socketChannel.pipeline()                                .addLast(new RpcDecoder(RpcRequest.class))                                .addLast(new RpcEncoder(RpcResponse.class))                                .addLast(new ServerRpcHandler(demoService));                    }                });        channel = serverBootstrap.bind("127.0.0.1",port).sync().channel();        logger.info("NettyRPC server listening on port "+ port + " and ready for connections...");        Runtime.getRuntime().addShutdownHook(new Thread(){            @Override            public void run(){                //do shutdown staff            }        });    }    public void stop() {        if (null == channel) {            throw new ServerStopException();        }        bossGroup.shutdownGracefully();        workerGroup.shutdownGracefully();        channel.closeFuture().syncUninterruptibly();        bossGroup = null;        workerGroup = null;        channel = null;    }}

• ServerRpcHandler

public class ServerRpcHandler extends SimpleChannelInboundHandler<RpcRequest> {    private static final Logger logger = LoggerFactory.getLogger(ServerRpcHandler.class);    private final Map<String, Object> serviceMapping;    public ServerRpcHandler(Map<String, Object> serviceMapping) {        this.serviceMapping = serviceMapping;    }    @Override    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RpcRequest rpcRequest) throws Exception {        RpcResponse response = new RpcResponse();        response.setTraceId(rpcRequest.getTraceId());        try {            logger.info("server handle request:{}",rpcRequest);            Object result = handle(rpcRequest);            response.setResult(result);        } catch (Throwable t) {            response.setError(t);        }        channelHandlerContext.writeAndFlush(response);    }    private Object handle(RpcRequest request) throws Throwable {        String className = request.getClassName();        Object serviceBean = serviceMapping.get(className);        Class<?> serviceClass = serviceBean.getClass();        String methodName = request.getMethodName();        Class<?>[] parameterTypes = request.getParameterTypes();        Object[] parameters = request.getParameters();        FastClass serviceFastClass = FastClass.create(serviceClass);        FastMethod serviceFastMethod = serviceFastClass.getMethod(methodName, parameterTypes);        return serviceFastMethod.invoke(serviceBean, parameters);    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {        logger.error(cause.getMessage(), cause);        RpcResponse response = new RpcResponse();        if(cause instanceof ServerException){            response.setTraceId(((ServerException) cause).getTraceId());        }        response.setError(cause);        ctx.writeAndFlush(response);    }}

• NettyClient

public class NettyClient implements IClient {    private EventLoopGroup workerGroup;    private Channel channel;    private int workerGroupThreads;    private ClientRpcHandler clientRpcHandler;    private final Optional<Pair<Long,TimeUnit>> NO_TIMEOUT = Optional.<Pair<Long,TimeUnit>>absent();    public NettyClient(int workerGroupThreads) {        this.workerGroupThreads = workerGroupThreads;    }    public void connect(InetSocketAddress socketAddress) {        workerGroup = new NioEventLoopGroup(workerGroupThreads);        clientRpcHandler = new ClientRpcHandler();        Bootstrap bootstrap = new Bootstrap();        bootstrap                .group(workerGroup)                .channel(NioSocketChannel.class)                .option(ChannelOption.SO_KEEPALIVE, true)                .option(ChannelOption.TCP_NODELAY, true)                .handler(new ChannelInitializer<SocketChannel>() {                    @Override                    protected void initChannel(SocketChannel ch) throws Exception {                        ch.pipeline()                                .addLast(new RpcDecoder(RpcResponse.class))                                .addLast(new RpcEncoder(RpcRequest.class))                                .addLast(clientRpcHandler);                    }                });        channel = bootstrap.connect(socketAddress.getAddress().getHostAddress(), socketAddress.getPort())                .syncUninterruptibly()                .channel();    }    public RpcResponse syncSend(RpcRequest request) throws InterruptedException {        System.out.println("send request:"+request);        channel.writeAndFlush(request).sync();        return clientRpcHandler.send(request,NO_TIMEOUT);    }    public RpcResponse asyncSend(RpcRequest request,TimeUnit timeUnit,long timeout) throws InterruptedException {        channel.writeAndFlush(request);        return clientRpcHandler.send(request, Optional.of(Pair.of(timeout,timeUnit)));    }    public InetSocketAddress getRemoteAddress() {        SocketAddress remoteAddress = channel.remoteAddress();        if (!(remoteAddress instanceof InetSocketAddress)) {            throw new RuntimeException("Get remote address error, should be InetSocketAddress");        }        return (InetSocketAddress) remoteAddress;    }    public void close() {        if (null == channel) {            throw new ClientCloseException();        }        workerGroup.shutdownGracefully();        channel.closeFuture().syncUninterruptibly();        workerGroup = null;        channel = null;    }}

• ClientRpcHandler

@ChannelHandler.Sharablepublic class ClientRpcHandler extends SimpleChannelInboundHandler<RpcResponse> {    //用blocking queue主要是用阻塞的功能，省的自己加锁    private final ConcurrentHashMap<String, BlockingQueue<RpcResponse>> responseMap = new ConcurrentHashMap<String, BlockingQueue<RpcResponse>>();    //messageReceived    @Override    protected void channelRead0(ChannelHandlerContext ctx, RpcResponse rpcResponse) throws Exception {        System.out.println("receive response:"+rpcResponse);        BlockingQueue<RpcResponse> queue = responseMap.get(rpcResponse.getTraceId());        queue.add(rpcResponse);    }    @Override    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {        super.exceptionCaught(ctx, cause);        cause.printStackTrace();    }    public RpcResponse send(RpcRequest request,Optional<Pair<Long,TimeUnit>> timeout) throws InterruptedException {        responseMap.putIfAbsent(request.getTraceId(), new LinkedBlockingQueue<RpcResponse>(1));        RpcResponse response = null;        try {            BlockingQueue<RpcResponse> queue = responseMap.get(request.getTraceId());            if(timeout == null || !timeout.isPresent()){                response = queue.take();            }else{                response = queue.poll(timeout.get().getKey(),timeout.get().getValue());            }        } finally {            responseMap.remove(request.getTraceId());        }        return response;    }}

• デコーダ

public class RpcDecoder extends ByteToMessageDecoder {    private Class<?> genericClass;    public RpcDecoder(Class<?> genericClass) {        this.genericClass = genericClass;    }    @Override    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {        if (byteBuf.readableBytes() < 4) {            return;        }        byteBuf.markReaderIndex();        int dataLength = byteBuf.readInt();        if (dataLength < 0) {            channelHandlerContext.close();        }        if (byteBuf.readableBytes() < dataLength) {            byteBuf.resetReaderIndex();        }        byte[] data = new byte[dataLength];        byteBuf.readBytes(data);        Object obj = SerializationUtil.deserializer(data, genericClass);        list.add(obj);    }}

• エンコーダ

public class RpcEncoder extends MessageToByteEncoder {    private Class<?> genericClass;    public RpcEncoder(Class<?> genericClass) {        this.genericClass = genericClass;    }    @Override    protected void encode(ChannelHandlerContext channelHandlerContext, Object obj, ByteBuf byteBuf) throws Exception {        if (genericClass.isInstance(obj)) {            byte[] data = SerializationUtil.serializer(obj);            byteBuf.writeInt(data.length);            byteBuf.writeBytes(data);        }    }}

プロジェクト ソース コード プロトコルデモ

リファレンス

• jvm-serializers

• protostuff

• java シリアル化/逆シリアル化における xstream、protobuf、protostuff の比較と使用例

• Protostuff のシリアル化

• protostuff の概要

• Protostuff の詳細説明

• シリアル化フレームワーク kryo VS hessian VS Protostuff VS java

• Protostuff のシリアル化と逆シリアル化

• eishay/jvm-serializers

• Protostuff シリアル化

• Netty を使用した多重化クライアントの実装