im-router 数据同步

im-common/src/main/java/net/sopod/soim/common/util/Jackson.java

@@ -176,6 +176,10 @@ public class Jackson {
 		return XML_INSTANCE;
 	}
 
+	public boolean canSerialize(Class<?> clazz) {
+		return objectMapper.canSerialize(clazz);
+	}
+
 	public <T> T deserialize(String content, Class<T> valueType) {
 		try {
 			return objectMapper.readValue(content, valueType);

im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSync.java

@@ -9,7 +9,10 @@ package net.sopod.soim.router.cache;
 public interface DataSync {
 
     /** 不是更新数据的方法开头 */
-    String[] nonUpdateMethodStart = new String[]{"get", "select", "list"};
+    String[] nonUpdateMethodStart = new String[]{"get", "find", "list",  "is", "select"};
+
+    /** 不是更新数据的方法 */
+    String[] nonUpdateMethod = new String[]{"toString", "equals", "hashCode", "wait", "getClass", "notify", "notifyAll"};
 
     /**
      * 如果是更新方法会同步数据，到新增节点或备份节点
@@ -20,6 +23,11 @@ public interface DataSync {
                 return false;
             }
         }
+        for (String method : nonUpdateMethod) {
+            if (method.equals(methodName)) {
+                return false;
+            }
+        }
         return true;
     }
 

im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSyncProxyFactory.java

@@ -4,11 +4,21 @@ import net.sf.cglib.proxy.Enhancer;
 import net.sf.cglib.proxy.MethodInterceptor;
 import net.sf.cglib.proxy.MethodProxy;
 import net.sopod.soim.common.util.Collects;
+import net.sopod.soim.common.util.Jackson;
+import net.sopod.soim.router.cache.annotation.Sync;
+import net.sopod.soim.router.cache.annotation.SyncIgnore;
+import net.sopod.soim.router.datasync.server.DataChangeTrigger;
+import net.sopod.soim.router.datasync.server.SyncTypes;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
 
 /**
  * BiSyncProxyManager
@@ -21,46 +31,139 @@ public class DataSyncProxyFactory {
 
     private static final Logger logger = LoggerFactory.getLogger(DataSyncProxyFactory.class);
 
+    /**
+     * 缓存数据类型更新方法列表
+     */
+    private static final Map<Class<? extends DataSync>, Set<String>> cacheTypeUpdaterMethods = new ConcurrentHashMap<>();
+
     @SuppressWarnings("unchecked")
-    public static <T extends DataSync> T newProxyInstance(Class<T> biSyncClazz) {
+    public static <T extends DataSync> T newProxyInstance(SyncTypes.SyncType<T> syncType) {
+        Class<T> type = syncType.dataType();
+        T instance;
+        try {
+            Constructor<T> constructor = type.getDeclaredConstructor();
+            constructor.setAccessible(true);
+            instance = constructor.newInstance();
+        } catch (InstantiationException | InvocationTargetException | NoSuchMethodException | IllegalAccessException e) {
+            throw new IllegalStateException(type + "实例创建失败,没有无参构造函数", e);
+        }
+
+        // 获取查询更新方法列表
+        Set<String> updaterMethods = cacheTypeUpdaterMethods.computeIfAbsent(type, dataType -> {
+            Method[] methods = dataType.getMethods();
+            Set<String> updaterMethodNames = new HashSet<>();
+            for (Method m : methods) {
+                SyncIgnore syncIgnore = m.getDeclaredAnnotation(SyncIgnore.class);
+                String methodName = m.getName();
+                if (syncIgnore != null) {
+                    continue;
+                }
+                boolean isUpdater = m.getDeclaringClass() != Object.class
+                        && instance.isUpdateMethod(methodName);
+                if (isUpdater) {
+                    if (updaterMethodNames.contains(methodName)) {
+                        throw new IllegalStateException(dataType + "重复的数据更新方法名" + methodName);
+                    }
+                    // 检查参数可序列化
+                    Class<?>[] paramTypes = m.getParameterTypes();
+                    for (int i = 0; i < paramTypes.length; i++) {
+                        if (!Jackson.json().canSerialize(paramTypes[i])) {
+                            throw new IllegalStateException(String.format("类:%s 更新方法:%s 第%di个参数不可序列化",
+                                    instance.getClass().getName(), methodName, i + 1));
+                        }
+                    }
+                    updaterMethodNames.add(methodName);
+                }
+            }
+            logger.info("{} 更新方法列表: {}", dataType, updaterMethodNames);
+            return updaterMethodNames;
+        });
+
+        // 创建代理对象
         Enhancer enhancer = new Enhancer();
-        enhancer.setSuperclass(biSyncClazz);
+        enhancer.setSuperclass(type);
-        enhancer.setCallback(new DataSyncProxyCallback());
+        enhancer.setCallback(new DataSyncProxyCallback<>(syncType, updaterMethods));
         return (T) enhancer.create();
     }
 
-    public static class DataSyncProxyCallback implements MethodInterceptor {
+    public static class DataSyncProxyCallback<T extends DataSync> implements MethodInterceptor {
+
+        private final SyncTypes.SyncType<T> syncType;
+
+        private final Set<String> updaterMethods;
+
+        public DataSyncProxyCallback(SyncTypes.SyncType<T> syncType, Set<String> updaterMethods) {
+            this.syncType = syncType;
+            this.updaterMethods = updaterMethods;
+        }
 
         @Override
         public Object intercept(Object instance, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
             String methodName = method.getName();
-            // 是判断是否更新的方法跳过
+            // 是判断不是更新的方法跳过
-            if ("isUpdateMethod".equals(methodName)) {
+            if (null != method.getDeclaredAnnotation(SyncIgnore.class)
+                    || !updaterMethods.contains(methodName)
+                    || "isUpdateMethod".equals(methodName)) {
                 return methodProxy.invokeSuper(instance, args);
             }
-            boolean isUpdateMethod = ((DataSync) instance).isUpdateMethod(methodName);
+
-            if (!isUpdateMethod) {
+            // TODO 记录更新操作(方法和参数)，查询数据订阅者，异步同步数据
+            DataChangeTrigger.instance().onUpdate(syncType, syncType.getDataKey((T) instance), methodName, args);
+
+            System.out.println("intercept invoke:" + methodName);
             return methodProxy.invokeSuper(instance, args);
         }
-            if (Collects.isNotEmpty(args)) {
+
-                // TODO 参数值可能为 null，检查参数可序列化放在生成代理对象时
-                for (int i = 0; i < args.length; i++) {
-                    if (!(args[i] instanceof Serializable)) {
-                        logger.error("类:{} 更新方法:{} 第{}i个参数不可序列化", instance.getClass(), methodName, i+1);
     }
+
+    public static interface A {
+        default void setName(String name) {
+            System.out.println("setName: " + name);
         }
+
     }
-            // TODO 记录更新操作(方法和参数)，查询数据订阅者，异步同步数据
+
-            System.out.println("intercept invoke:" + methodName);
+    public static class B implements A {
-            return methodProxy.invokeSuper(instance, args);
+        private int age;
+
+        public void setAge(int age) {
+            this.age = age;
         }
 
+        @Override
+        public void setName(String name) {
+            System.out.println("over setName: " + name);
+        }
     }
 
     public static void main(String[] args) {
-        RouterUser routerUser = newProxyInstance(RouterUser.class);
+
+        RouterUser routerUser = newProxyInstance(SyncTypes.ROUTER_USER);
         routerUser.setAccount("日月光");
         System.out.println(routerUser.getAccount());
+
+//        Enhancer enhancer = new Enhancer();
+//        enhancer.setSuperclass(B.class);
+//        enhancer.setCallback(new MethodInterceptor() {
+//            @Override
+//            public Object intercept(Object o, Method method, Object[] args, MethodProxy methodProxy) throws Throwable {
+//                System.out.println("proxy method: " + method.getName());
+//                return methodProxy.invokeSuper(o, args);
+//            }
+//        });
+//
+//        B b = (B) enhancer.create();
+//        b.setAge(12);
+//        b.setName("沧海");
+//
+//        System.out.println(b.age);
+//
+//        for (Method method : B.class.getMethods()) {
+//            System.out.println(method.getName() + "-" + method.hashCode() + ": " + method.getDeclaringClass());
+//        }
+//        System.out.println(Serializable.class.isAssignableFrom(String.class));
+//        System.out.println(Serializable.class.isAssignableFrom(Integer.class));
+//        System.out.println(Integer.class.isAssignableFrom(Serializable.class));
     }
 
 }

im-service/im-router/src/main/java/net/sopod/soim/router/cache/SoImUserCache.java

@@ -1,5 +1,8 @@
 package net.sopod.soim.router.cache;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
@@ -12,6 +15,8 @@ import java.util.concurrent.ConcurrentHashMap;
  */
 public class SoImUserCache {
 
+    private static final Logger logger = LoggerFactory.getLogger(SoImUserCache.class);
+
     private static final SoImUserCache INSTANCE = new SoImUserCache();
 
     private final ConcurrentHashMap<Long, RouterUser> routerUserMap = new ConcurrentHashMap<>(128);
@@ -20,14 +25,18 @@ public class SoImUserCache {
         return INSTANCE;
     }
 
-    public void put(Long uid, RouterUser routerUser) {
+    public RouterUser put(Long uid, RouterUser routerUser) {
-        routerUserMap.put(uid, routerUser);
+        return routerUserMap.put(uid, routerUser);
     }
 
     public RouterUser get(Long uid) {
         return routerUserMap.get(uid);
     }
 
+    public RouterUser remove(Long uid) {
+        return routerUserMap.remove(uid);
+    }
+
     public Map<Long, RouterUser> getRouterUserMap() {
         return routerUserMap;
     }

im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/Sync.java

@@ -0,0 +1,19 @@
+package net.sopod.soim.router.cache.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * Sync
+ * 修改数据的方法操作
+ *
+ * @author tmy
+ * @date 2022-05-05 17:18
+ */
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD})
+public @interface Sync {
+
+
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/SyncIgnore.java

@@ -1,5 +1,7 @@
 package net.sopod.soim.router.cache.annotation;
 
+import java.lang.annotation.*;
+
 /**
  * BiSyncIgnore
  * 忽略方法同步
@@ -7,6 +9,9 @@ package net.sopod.soim.router.cache.annotation;
  * @author tmy
  * @date 2022-05-04 17:39
  */
-public @interface DataSyncIgnore {
+@Documented
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.METHOD})
+public @interface SyncIgnore {
 
 }

im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java

@@ -81,6 +81,7 @@ public class ImRouterAppOnReady implements ApplicationListener<ApplicationReadyE
      * 2.计算所有需要迁移数据的节点，当前虚拟节点的顺时针临近节点
      * 3.依次调用数据迁移节点，分页拉取router缓存用户数据（如有数据节点获取失败，探测是否可用，不可用从第1步重新开始；）
      *   如已拉取数据有更新需要双写？？[CGLib], 服务注册前需要被调用，开netty http服务与dubbo服务端口偏移量1000
+     *   启用数据压缩，可能遇到大对象参数等
      * 4.数据迁移完成后，注册dubbo服务
      * 5.定时n秒后或添加provider filter服务第一次被调用m秒后，依次调用所有迁移数据节点可清空用户数据（用户数据已移过来了失败不用管）
      *   可能依次调用清空数据过程中又会有数据更新需要订阅

im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/DataChangeTrigger.java

@@ -0,0 +1,42 @@
+package net.sopod.soim.router.datasync.server;
+
+import net.sopod.soim.router.cache.DataSync;
+
+import java.util.Queue;
+import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * DataUpdateTrigger
+ * 序列化后，放入处理异步队列发送给监听者
+ *
+ * @author tmy
+ * @date 2022-05-05 16:14
+ */
+public class DataChangeTrigger {
+
+    private static final DataChangeTrigger INSTANCE = new DataChangeTrigger();
+
+    public static DataChangeTrigger instance() {
+        return INSTANCE;
+    }
+
+    private AtomicInteger seqCounter = new AtomicInteger();
+
+    private Queue<SyncLog> logQueue = new ConcurrentLinkedQueue<>();
+
+    public <T extends DataSync> void onUpdate(SyncTypes.SyncType<T> syncType, String dataKey, String method, Object[] args) {
+        // 序列化 args，避免后续更改
+        // logQueue.add()
+    }
+
+    public <T extends DataSync> void onAdd(SyncTypes.SyncType<T> syncType, T data) {
+        // 序列化 data，避免后续更改
+
+    }
+
+    public <T extends DataSync> void onRemove(SyncTypes.SyncType<T> syncType, String dataKey) {
+
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncClient.java

@@ -0,0 +1,30 @@
+package net.sopod.soim.router.datasync.server;
+
+import io.netty.bootstrap.Bootstrap;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+
+/**
+ * SyncClient
+ *
+ * @author tmy
+ * @date 2022-05-05 15:03
+ */
+public class SyncClient {
+
+    public SyncClient() {
+        NioEventLoopGroup group = new NioEventLoopGroup(2);
+        new Bootstrap()
+                .group(group)
+                .handler(new ChannelInitializer<SocketChannel>() {
+                    @Override
+                    protected void initChannel(SocketChannel channel) throws Exception {
+                        channel.pipeline()
+                            .addLast(new SyncLogDataCodec());
+                    }
+                });
+    }
+
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncLog.java

@@ -0,0 +1,200 @@
+package net.sopod.soim.router.datasync.server;
+
+import io.netty.buffer.ByteBuf;
+import lombok.Data;
+import lombok.experimental.Accessors;
+import net.sopod.soim.common.util.Jackson;
+import org.apache.dubbo.common.io.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+import java.io.Serializable;
+import java.lang.reflect.Method;
+import java.nio.charset.StandardCharsets;
+
+/**
+ * SyncLog
+ * syncLog serialize
+ *
+ * @author tmy
+ * @date 2022-05-05 11:37
+ */
+@Data
+@Accessors(chain = true)
+public class SyncLog implements Serializable {
+
+    private static final Logger logger = LoggerFactory.getLogger(SyncLog.class);
+
+    private static final long serialVersionUID = -8925525709590423526L;
+
+    private static final short MAGIC = 0x7a21;
+
+    public static final int OPT_ADD = 1;
+    public static final int OPT_REMOVE = 2;
+    public static final int OPT_UPDATE = 3;
+
+    /** 日志序列号保证顺序 */
+    private int logSeq;
+
+    /**
+     * 操作类型:
+     *   1.新增
+     *   2.删除
+     *   3.更新
+     */
+    private int operateType;
+
+    /**
+     * {@link SyncTypes} ordinal
+     */
+    private int syncDataType;
+
+    /** 数据id标示 */
+    private String dataKey;
+
+    /** ================ 新增参数:序列化后的数据 ===================== */
+    private String addSerializeData;
+
+    /** ================ 更新参数 ===================== */
+    private String clazz;
+
+    private String method;
+
+    private Object[] args;
+
+    public byte[] toBytes() {
+        byte[] clazzBytes = clazz.getBytes();
+        byte[] methodBytes = method.getBytes();
+        int argSize = args == null ? 0 : args.length;
+
+        byte[][] byteArgs = new byte[argSize][];
+        if (args != null) {
+            for (int i = 0; i < args.length; i++) {
+                // 反序列化时根据方法参数类型json反序列化
+                // TODO null
+                String argJson = Jackson.json().serialize(args[i]);
+                byteArgs[i] = argJson.getBytes(StandardCharsets.UTF_8);
+            }
+        }
+        int argsByteLen = 0;
+        for (byte[] byteArg : byteArgs) {
+            argsByteLen += 4;
+            argsByteLen += byteArg.length;
+        }
+        // 总长 + 同步数据类型(byte) +
+        byte[] bytes = new byte[
+                2   // 魔术
+                + 4 // 请求体总长度
+                + 1 // 同步数据类型
+                + 4 // clazz 字节长度
+                + clazzBytes.length // clazz字节
+                + 4 // method 字节长度
+                + methodBytes.length // method 字节
+                + 4 // args参数个数
+                + argsByteLen // args参数字节
+        ];
+        int offset = 0;
+        Bytes.short2bytes(MAGIC, bytes, offset);
+        offset += 2;
+
+        Bytes.int2bytes(bytes.length - 6, bytes, offset);
+        offset += 4;
+
+        bytes[offset] = (byte)syncDataType;
+        offset += 1;
+
+        Bytes.int2bytes(clazzBytes.length, bytes, offset);
+        offset += 4;
+        System.arraycopy(clazzBytes, 0, bytes, offset, clazzBytes.length);
+        offset += clazzBytes.length;
+
+        Bytes.int2bytes(methodBytes.length, bytes, offset);
+        offset += 4;
+        System.arraycopy(methodBytes, 0, bytes, offset, methodBytes.length);
+        offset += methodBytes.length;
+
+        Bytes.int2bytes(argSize, bytes, offset);
+        offset += 4;
+        if (argsByteLen > 0) {
+            for (byte[] byteArg : byteArgs) {
+                Bytes.int2bytes(byteArg.length, bytes, offset);
+                offset += 4;
+                System.arraycopy(byteArg, 0, bytes, offset, byteArg.length);
+                offset += byteArg.length;
+            }
+        }
+        return bytes;
+    }
+
+    public static SyncLog read(ByteBuf buf) {
+        short magic = buf.readShort();
+        if (magic != MAGIC) {
+            throw new IllegalStateException("unknown bytes magic error");
+        }
+        SyncLog log = new SyncLog();
+        // 后续bytes长度
+        int dataLen = buf.readInt();
+        log.syncDataType = buf.readByte();
+        int clazzLen = buf.readInt();
+        byte[] clazzBytes = new byte[clazzLen];
+        buf.readBytes(clazzBytes);
+        log.clazz = new String(clazzBytes, StandardCharsets.UTF_8);
+        int methodLen = buf.readInt();
+        byte[] methodBytes = clazzLen >= methodLen ? clazzBytes : new byte[methodLen];
+        buf.readBytes(methodBytes, 0, methodLen);
+        log.method = new String(methodBytes, 0, methodLen, StandardCharsets.UTF_8);
+        int argSize = buf.readInt();
+        log.args = new Object[argSize];
+        if (argSize > 0) {
+            Method method = getClassMethod(log.clazz, log.method);
+            if (method == null) {
+                throw new IllegalStateException("类" + log.clazz + "方法" + log.method + "未找到");
+            }
+            Class<?>[] paramTypes = method.getParameterTypes();
+            if (paramTypes.length != argSize) {
+                throw new IllegalStateException("类" + log.clazz + "方法" + log.method + "指定参数" + argSize + "个,查到参数" + paramTypes.length + "个");
+            }
+            // 复用 bytes
+            byte[] argBytes = new byte[0];
+            for (int i = 0; i < argSize; i++) {
+                int argLen = buf.readInt();
+                argBytes = argBytes.length >= argLen ? argBytes : new byte[argLen];
+                buf.readBytes(argBytes, 0, argLen);
+                String argJson = new String(argBytes, 0, argLen, StandardCharsets.UTF_8);
+                Object arg = Jackson.json().deserialize(argJson, paramTypes[i]);
+                log.args[i] = arg;
+            }
+        }
+        return log;
+    }
+
+    /**
+     * TODO 缓存反射结果
+     */
+    @Nullable
+    private static Method getClassMethod(String clazzName, String methodName) {
+        try {
+            Class<?> clazz = Class.forName(clazzName);
+            Method[] methods = clazz.getDeclaredMethods();
+            for (Method method : methods) {
+                if (method.getName().equals(methodName)) {
+                    return method;
+                }
+            }
+        } catch (ClassNotFoundException e) {
+            logger.error("获取类型方法失败: {}.{}", clazzName, methodName, e);
+        }
+        return null;
+    }
+
+    public static void main(String[] args) {
+        new SyncLog()
+                .setSyncDataType(SyncTypes.ROUTER_USER.ordinal())
+                .setClazz("")
+                .setMethod("")
+                .setArgs(new Object[]{});
+
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncLogDataCodec.java

@@ -0,0 +1,50 @@
+package net.sopod.soim.router.datasync.server;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.*;
+import io.netty.handler.codec.ByteToMessageDecoder;
+import io.netty.handler.codec.MessageToByteEncoder;
+
+import java.util.List;
+
+/**
+ * SyncDataInboundHandler
+ *
+ * @author tmy
+ * @date 2022-05-05 10:20
+ */
+public class SyncLogDataCodec
+        extends CombinedChannelDuplexHandler<SyncLogDataCodec.SyncDataDecoder,
+        SyncLogDataCodec.SyncDataEncoder> {
+
+    public SyncLogDataCodec() {
+        super(new SyncDataDecoder(), new SyncDataEncoder());
+    }
+
+    /**
+     * 解码器
+     */
+    public static class SyncDataDecoder extends ByteToMessageDecoder {
+
+        @Override
+        protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
+            SyncLog syncLog = SyncLog.read(byteBuf);
+            list.add(syncLog);
+        }
+
+    }
+
+    /**
+     * 编码器
+     */
+    public static class SyncDataEncoder extends MessageToByteEncoder<SyncLog> {
+
+        @Override
+        protected void encode(ChannelHandlerContext channelHandlerContext, SyncLog syncLog, ByteBuf byteBuf) throws Exception {
+            byte[] bytes = syncLog.toBytes();
+            byteBuf.writeBytes(bytes);
+        }
+
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncLogInboundHandler.java

@@ -0,0 +1,19 @@
+package net.sopod.soim.router.datasync.server;
+
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+
+/**
+ * SyncLogInboundHandler
+ *
+ * @author tmy
+ * @date 2022-05-05 14:57
+ */
+public class SyncLogInboundHandler extends SimpleChannelInboundHandler<SyncLog> {
+
+    @Override
+    protected void channelRead0(ChannelHandlerContext channelHandlerContext, SyncLog syncLog) throws Exception {
+        
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncServer.java

@@ -0,0 +1,50 @@
+package net.sopod.soim.router.datasync.server;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.ChannelPipeline;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import io.netty.handler.logging.LogLevel;
+import io.netty.handler.logging.LoggingHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * SyncServer
+ *
+ * @author tmy
+ * @date 2022-05-05 10:02
+ */
+public class SyncServer {
+
+    private static final Logger logger = LoggerFactory.getLogger(SyncServer.class);
+
+    public SyncServer() {
+        NioEventLoopGroup boss = new NioEventLoopGroup(1);
+        NioEventLoopGroup worker = new NioEventLoopGroup(2);
+        ServerBootstrap serverBoot = new ServerBootstrap()
+                .group(boss, worker)
+                .channel(NioServerSocketChannel.class)
+                .childHandler(new ChannelInitializer<>() {
+                    @Override
+                    protected void initChannel(Channel channel) throws Exception {
+                        LogLevel logLevel = logger.isDebugEnabled() ? LogLevel.DEBUG
+                                : logger.isInfoEnabled() ? LogLevel.INFO
+                                : logger.isWarnEnabled() ? LogLevel.WARN
+                                : logger.isErrorEnabled() ? LogLevel.ERROR : LogLevel.INFO;
+                        ChannelPipeline pipeline = channel.pipeline();
+                        pipeline.addLast(new LoggingHandler(logLevel))
+                                .addLast(new SyncLogDataCodec())
+                                .addLast(new SyncLogInboundHandler());
+                    }
+                });
+        serverBoot.bind(8080);
+    }
+
+    public void start() {
+
+    }
+
+}

im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncTypes.java

@@ -0,0 +1,126 @@
+package net.sopod.soim.router.datasync.server;
+
+import net.sopod.soim.common.util.StringUtil;
+import net.sopod.soim.router.cache.DataSync;
+import net.sopod.soim.router.cache.RouterUser;
+import net.sopod.soim.router.cache.SoImUserCache;
+
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * SyncTypeEnum
+ * 需要同步的数据类型
+ *
+ * @author tmy
+ * @date 2022-05-05 10:40
+ */
+public class SyncTypes {
+
+    /**
+     * 模拟枚举根据顺序值获取数据类型对象
+     */
+    public static <T extends DataSync> SyncType<T> getSyncType(int ordinal) {
+        return SyncType.getSyncType(ordinal);
+    }
+
+    /**
+     * 用户数据同步处理:
+     * 改: 代理对象方法监控(getData(dataKey), updateMethod(args))
+     * 增: 增数据(data)
+     * 删: 删数据(dataKey)
+     */
+    public static final SyncType<RouterUser> ROUTER_USER = new SyncType<>(RouterUser.class, "ROUTER_USER") {
+        @Override
+        public String getDataKey(RouterUser data) {
+            return StringUtil.toString(data.getUid());
+        }
+
+        @Override
+        public RouterUser getData(String uid) {
+            return SoImUserCache.getInstance().get(Long.valueOf(uid));
+        }
+
+        @Override
+        public boolean addData(RouterUser data) {
+            SoImUserCache.getInstance().put(data.getUid(), data);
+            return true;
+        }
+
+        @Override
+        public boolean removeData(String uid) {
+            return null != SoImUserCache.getInstance().remove(Long.valueOf(uid));
+        }
+    };
+
+    public static abstract class SyncType<T extends DataSync> {
+        private static final List<SyncType<?>> TYPES = new ArrayList<>();
+        private static int ordinalSeq = Byte.MIN_VALUE;
+
+        private final int ordinal;
+        private final Class<T> dataType;
+        private final String name;
+
+        SyncType(Class<T> dataType, String name) {
+            this.ordinal = ordinalSeq++;
+            if (ordinal > Byte.MAX_VALUE) {
+                throw new IllegalStateException("类型超过255个,需修改协议");
+            }
+            this.dataType = dataType;
+            this.name = name;
+            TYPES.add(this);
+        }
+
+        public Class<T> dataType() {
+            return dataType;
+        }
+
+        public String name() {
+            return name;
+        }
+
+        public int ordinal() {
+            return ordinal;
+        }
+
+        public abstract String getDataKey(T data);
+
+        public abstract T getData(String key);
+
+        public abstract boolean addData(T data);
+
+        public abstract boolean removeData(String key);
+
+        @Nullable
+        @SuppressWarnings("unchecked")
+        static <T extends DataSync> SyncType<T> getSyncType(int ordinal) {
+            return (SyncType<T>) (ordinal < TYPES.size() ? TYPES.get(ordinal) : null);
+        }
+
+        @Override
+        public String toString() {
+            return "SyncType{" +
+                    "ordinal=" + ordinal +
+                    ", dataType=" + dataType +
+                    ", name='" + name + '\'' +
+                    '}' + "@" + Integer.toHexString(this.hashCode());
+        }
+
+    }
+
+    public static void main(String[] args) {
+        System.out.println(ROUTER_USER.ordinal());
+        System.out.println(SyncTypes.getSyncType(0));
+        System.out.println(SyncTypes.getSyncType(1));
+        System.out.println(SyncTypes.getSyncType(2));
+        Class<RouterUser> routerUserClass = SyncTypes.ROUTER_USER.dataType();
+
+        ConcurrentHashMap<String, String> map = new ConcurrentHashMap<>();
+        map.put("1", "2");
+        System.out.println(map.remove("2"));
+        System.out.println(map.remove("1"));
+    }
+
+}