netty serializer

Author: xlorne

tx-commons/src/main/java/com/codingapi/txlcn/commons/util/serializer/ISerializer.java

@@ -18,18 +18,22 @@
 
 import com.codingapi.txlcn.commons.exception.SerializerException;
 
+import java.io.InputStream;
+import java.io.OutputStream;
+
 /**
  * @author lorne 2017/11/11
  */
 public interface ISerializer {
+
     /**
      * 序列化对象
      *
      * @param obj 需要序更列化的对象
-     * @return byte []  序列号结果
+     * @param outputStream  写入对象
      * @throws SerializerException  序列化异常
      */
-    byte[] serialize(Object obj) throws SerializerException;
+    void serialize(Object obj, OutputStream outputStream) throws SerializerException ;
 
 
     /**
@@ -43,4 +47,30 @@ public interface ISerializer {
      */
 
     <T> T deSerialize(byte[] param, Class<T> clazz) throws SerializerException;
+
+
+
+
+    /**
+     * 反序列化对象
+     *
+     * @param inputStream 需要反序列化的inputStream
+     * @param clazz 反序列化成为的bean对象Class
+     * @param <T>   反序列化成为的bean对象
+     * @return  对象
+     * @throws SerializerException  序列化异常
+     */
+
+    <T> T deSerialize(InputStream inputStream, Class<T> clazz) throws SerializerException;
+
+
+    /**
+     * 序列化对象
+     *
+     * @param obj 需要序更列化的对象
+     * @return byte []  序列号结果
+     * @throws SerializerException  序列化异常
+     */
+    byte[] serialize(Object obj) throws SerializerException;
+
 }

tx-commons/src/main/java/com/codingapi/txlcn/commons/util/serializer/ProtostuffSerializer.java

@@ -22,8 +22,7 @@
 import org.objenesis.Objenesis;
 import org.objenesis.ObjenesisStd;
 
-import java.io.ByteArrayInputStream;
-import java.io.ByteArrayOutputStream;
+import java.io.*;
 
 /**
  * @author lorne 2017/11/11
@@ -54,6 +53,22 @@ public byte[] serialize(Object obj) throws SerializerException {
         }
     }
 
+
+    @Override
+    public void serialize(Object obj, OutputStream outputStream) throws SerializerException {
+        Class cls = obj.getClass();
+        LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
+        try {
+            Schema schema = getSchema(cls);
+            ProtostuffIOUtil.writeTo(outputStream, obj, schema, buffer);
+        } catch (Exception e) {
+            throw new SerializerException(e.getMessage(), e);
+        }finally {
+            buffer.clear();
+        }
+    }
+
+
     @Override
     public <T> T deSerialize(byte[] param, Class<T> cls) throws SerializerException {
         T object;
@@ -66,5 +81,18 @@ public <T> T deSerialize(byte[] param, Class<T> cls) throws SerializerException
             throw new SerializerException(e.getMessage(), e);
         }
     }
+
+    @Override
+    public <T> T deSerialize(InputStream inputStream, Class<T> cls) throws SerializerException {
+        T object;
+        try{
+            object = OBJENESIS.newInstance(cls);
+            Schema schema = getSchema(cls);
+            ProtostuffIOUtil.mergeFrom(inputStream, object, schema);
+            return object;
+        } catch (Exception e) {
+            throw new SerializerException(e.getMessage(), e);
+        }
+    }
 }
 

tx-commons/src/main/java/com/codingapi/txlcn/commons/util/serializer/SerializerContext.java

@@ -17,6 +17,9 @@
 
 import com.codingapi.txlcn.commons.exception.SerializerException;
 
+import java.io.InputStream;
+import java.io.OutputStream;
+
 /**
  * @author lorne  2018/12/31
  *
@@ -52,4 +55,14 @@ public byte[] serialize(Object obj) throws SerializerException {
     public <T> T deSerialize(byte[] param, Class<T> clazz) throws SerializerException {
         return protostuffSerializer.deSerialize(param,clazz);
     }
+
+    @Override
+    public <T> T deSerialize(InputStream inputStream, Class<T> clazz) throws SerializerException {
+        return protostuffSerializer.deSerialize(inputStream,clazz);
+    }
+
+    @Override
+    public void serialize(Object obj, OutputStream outputStream) throws SerializerException {
+        protostuffSerializer.serialize(obj, outputStream);
+    }
 }

tx-spi-message-netty/src/main/java/com/codingapi/txlcn/spi/message/netty/bean/RpcCmdContext.java

@@ -34,7 +34,9 @@
 @Slf4j
 public class RpcCmdContext {
 
-    private RpcConfig rpcConfig;
+    private int cacheSize = 1024;
+
+    private int waitTime = 1;
 
     private static RpcCmdContext context = null;
 
@@ -108,7 +110,7 @@ private RpcContent findRpcContent() {
     }
 
     private RpcContent createRpcContent() {
-        if (cacheList.size() < rpcConfig.getCacheSize()) {
+        if (cacheList.size() < cacheSize) {
             RpcContent rpcContent = new RpcContent(getWaitTime());
             rpcContent.init();
             cacheList.add(rpcContent);
@@ -136,10 +138,11 @@ private void clearKey(String key) {
 
 
     public void setRpcConfig(RpcConfig rpcConfig) {
-        this.rpcConfig = rpcConfig;
+        cacheSize = rpcConfig.getCacheSize();
+        waitTime = (int) rpcConfig.getWaitTime()/1000;
     }
 
     public int getWaitTime() {
-        return (int) (rpcConfig.getWaitTime() / 1000);
+        return waitTime;
     }
 }

tx-spi-message-netty/src/main/java/com/codingapi/txlcn/spi/message/netty/bean/SocketManager.java

@@ -50,7 +50,7 @@ public class SocketManager {
 
     private static SocketManager manager = null;
 
-    private RpcConfig rpcConfig;
+    private long attrDelayTime = 1000*60;
 
     private SocketManager() {
         channels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
@@ -89,7 +89,7 @@ public void removeChannel(Channel channel) {
             executorService.schedule(() -> {
                 String key = channel.remoteAddress().toString();
                 appNames.remove(key);
-            }, rpcConfig.getAttrDelayTime(), TimeUnit.MILLISECONDS);
+            }, attrDelayTime, TimeUnit.MILLISECONDS);
         } catch (RejectedExecutionException ignored) {
             // caused down server.
         }
@@ -182,7 +182,7 @@ public void bindModuleName(String remoteKey, String moduleName) {
     }
 
     public void setRpcConfig(RpcConfig rpcConfig) {
-        this.rpcConfig = rpcConfig;
+        attrDelayTime = rpcConfig.getAttrDelayTime();
     }
 
     /**

tx-spi-message-netty/src/main/java/com/codingapi/txlcn/spi/message/netty/handler/NettyRpcClientHandlerInitHandler.java

@@ -19,9 +19,6 @@
 import io.netty.channel.ChannelInitializer;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import io.netty.handler.codec.LengthFieldPrepender;
-import io.netty.handler.codec.serialization.ClassResolvers;
-import io.netty.handler.codec.serialization.ObjectDecoder;
-import io.netty.handler.codec.serialization.ObjectEncoder;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
@@ -43,11 +40,14 @@ public class NettyRpcClientHandlerInitHandler extends ChannelInitializer<Channel
 
     @Override
     protected void initChannel(Channel ch) throws Exception {
+
         ch.pipeline().addLast(new LengthFieldPrepender(4, false));
-        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE, 0, 4, 0, 4));
+        ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(Integer.MAX_VALUE,
+                0, 4, 0, 4));
+
+        ch.pipeline().addLast(new ObjectSerializerEncoder());
+        ch.pipeline().addLast(new ObjectSerializerDecoder());
 
-        ch.pipeline().addLast(new ObjectDecoder(ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
-        ch.pipeline().addLast(new ObjectEncoder());
 
         ch.pipeline().addLast(new RpcCmdDecoder());
         ch.pipeline().addLast(new RpcCmdEncoder());

tx-spi-message-netty/src/main/java/com/codingapi/txlcn/spi/message/netty/handler/NettyRpcServerHandlerInitHandler.java

@@ -20,9 +20,6 @@
 import io.netty.channel.ChannelInitializer;
 import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
 import io.netty.handler.codec.LengthFieldPrepender;
-import io.netty.handler.codec.serialization.ClassResolvers;
-import io.netty.handler.codec.serialization.ObjectDecoder;
-import io.netty.handler.codec.serialization.ObjectEncoder;
 import io.netty.handler.timeout.IdleStateHandler;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -56,8 +53,10 @@ protected void initChannel(Channel ch) throws Exception {
         ch.pipeline().addLast(new IdleStateHandler(managerProperties.getCheckTime(),
                 managerProperties.getCheckTime(), managerProperties.getCheckTime(), TimeUnit.MILLISECONDS));
 
-        ch.pipeline().addLast(new ObjectDecoder(ClassResolvers.weakCachingConcurrentResolver(this.getClass().getClassLoader())));
-        ch.pipeline().addLast(new ObjectEncoder());
+
+        ch.pipeline().addLast(new ObjectSerializerEncoder());
+        ch.pipeline().addLast(new ObjectSerializerDecoder());
+
 
         ch.pipeline().addLast(new RpcCmdDecoder());
         ch.pipeline().addLast(new RpcCmdEncoder());

tx-spi-message-netty/src/main/java/com/codingapi/txlcn/spi/message/netty/handler/ObjectSerializerDecoder.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2017-2019 CodingApi .
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.codingapi.txlcn.spi.message.netty.handler;
+
+import com.codingapi.txlcn.commons.util.serializer.SerializerContext;
+import com.codingapi.txlcn.spi.message.netty.bean.NettyRpcCmd;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufInputStream;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToMessageDecoder;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.List;
+
+/**
+ * @author lorne
+ * @date 2019/1/19
+ * @description
+ */
+@Slf4j
+public class ObjectSerializerDecoder extends MessageToMessageDecoder<ByteBuf> {
+
+
+    @Override
+    protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception {
+        if(msg==null){
+            return;
+        }
+        ByteBufInputStream byteBufInputStream =  new ByteBufInputStream(msg, true);
+        NettyRpcCmd object = SerializerContext.getInstance().deSerialize(byteBufInputStream, NettyRpcCmd.class);
+        out.add(object);
+    }
+
+
+}

tx-spi-message-netty/src/main/java/com/codingapi/txlcn/spi/message/netty/handler/ObjectSerializerEncoder.java

@@ -0,0 +1,43 @@
+/*
+ * Copyright 2017-2019 CodingApi .
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.codingapi.txlcn.spi.message.netty.handler;
+
+import com.codingapi.txlcn.commons.util.serializer.SerializerContext;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.ByteBufOutputStream;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.MessageToByteEncoder;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.Serializable;
+
+/**
+ * @author lorne
+ * @date 2019/1/19
+ * @description
+ */
+@Slf4j
+public class ObjectSerializerEncoder extends MessageToByteEncoder<Serializable> {
+
+
+    @Override
+    protected void encode(ChannelHandlerContext ctx, Serializable msg, ByteBuf out) throws Exception {
+        ByteBufOutputStream bout = new ByteBufOutputStream(out);
+        SerializerContext.getInstance().serialize(msg,bout);
+    }
+
+
+}

tx-spi-message-netty/src/main/java/com/codingapi/txlcn/spi/message/netty/impl/NettyRpcClientInitializer.java

@@ -84,7 +84,7 @@ public synchronized void connect(SocketAddress socketAddress) {
                     b.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 5000);
                     b.handler(nettyRpcClientHandlerInitHandler);
                     ChannelFuture channelFuture = b.connect(socketAddress).syncUninterruptibly();
-                    log.info("client -> {} , transactionState:{}", socketAddress, channelFuture.isSuccess());
+                    log.info("client -> {} , state :{}", socketAddress, channelFuture.isSuccess());
                     connected = true;
                     break;