im-router数据同步服务端客户端

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

@@ -72,6 +72,36 @@ public class ImRouterAppOnReady implements ApplicationListener<ApplicationReadyE
         return Ordered.HIGHEST_PRECEDENCE;
     }
 
+    /**
+     * 检查集群情况
+     */
+    public void checkClusterEnvironment() {
+        List<Instance> clusterImEntryInstance = getClusterImEntryInstance();
+        if (Collects.isEmpty(clusterImEntryInstance)) {
+            logger.info("当前集群无{}节点，直接启动", AppConstant.APP_IM_ROUTER_NAME);
+            return;
+        }
+        for (Instance instance : clusterImEntryInstance) {
+            String host = instance.getIp();
+            // 同步服务器端口偏移量 1000
+            int port = instance.getPort();
+        }
+    }
+
+    private List<Instance> getClusterImEntryInstance() {
+        String serverAddr = "124.222.131.236:3848";
+        Properties properties = new Properties();
+        properties.put("serverAddr", serverAddr);
+        // 获取当前服务实例
+        NamingService namingService = null;
+        try {
+            namingService = NacosFactory.createNamingService(properties);
+            return namingService.getAllInstances(AppConstant.APP_IM_ROUTER_NAME);
+        } catch (NacosException e) {
+            throw new IllegalStateException("集群状态检查失败:", e);
+        }
+    }
+
     /**
      * 一致性hash，使用了虚拟节点会导致迁移多个数据节点
      *

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

@@ -4,6 +4,7 @@ import net.sopod.soim.common.util.Jackson;
 import net.sopod.soim.router.cache.RouterUser;
 import net.sopod.soim.router.cache.RouterUserStorage;
 
+import java.nio.channels.Channel;
 import java.util.Map;
 
 /**
@@ -14,6 +15,10 @@ import java.util.Map;
  */
 public class SyncService {
 
+    public SyncService(String clientAddr, Channel channel) {
+
+    }
+
     public void fullSync() {
         Map<Long, RouterUser> routerUserMap = RouterUserStorage.getInstance().getRouterUserMap();
         for (Map.Entry<Long, RouterUser> entry : routerUserMap.entrySet()) {

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

@@ -5,7 +5,10 @@ import io.netty.channel.Channel;
 import io.netty.channel.ChannelInitializer;
 import io.netty.channel.nio.NioEventLoopGroup;
 import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioSocketChannel;
 import io.netty.util.AttributeKey;
+import net.sopod.soim.router.cache.RouterUser;
+import net.sopod.soim.router.datasync.SyncTypes;
 import net.sopod.soim.router.datasync.server.codec.SyncCmdCodec;
 import net.sopod.soim.router.datasync.server.codec.SyncLogEncoder;
 import net.sopod.soim.router.datasync.server.data.SyncCmd;
@@ -29,6 +32,7 @@ public class SyncClient {
 
     public SyncClient() {
         this.bootstrap = new Bootstrap()
+                .channel(NioSocketChannel.class)
                 .handler(new ChannelInitializer<SocketChannel>() {
                     @Override
                     protected void initChannel(SocketChannel channel) throws Exception {
@@ -69,4 +73,30 @@ public class SyncClient {
         this.group.shutdownGracefully();
     }
 
+
+    public static void main(String[] args) throws InterruptedException {
+        SyncClient client = new SyncClient();
+        client.connect("127.0.0.1", 9999);
+        // SyncCmd syncCmd = new SyncCmd().setCmdType(SyncCmd.SYNC_LOG);
+        // new SyncLog()
+        RouterUser user = new RouterUser()
+                .setUid(12312L)
+                .setAccount("蓝水云烟")
+                .setImEntryAddr("127.0.0.1")
+                .setIsOnline(false)
+                .setOnlineTime(10086L);
+        RouterUser user2 = new RouterUser()
+                .setUid(10010L)
+                .setAccount("百战成诗")
+                .setImEntryAddr("192.168.1.101")
+                .setIsOnline(true)
+                .setOnlineTime(16161L);
+        SyncLog.AddLog<RouterUser> addLog = SyncLog.addLog(1, SyncTypes.ROUTER_USER)
+                .addData(user)
+                .addData(user2);
+        client.clientChannel.writeAndFlush(addLog);
+
+        Thread.sleep(10000);
+        client.close();
+    }
 }

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

@@ -58,8 +58,8 @@ public class SyncLog implements Serializable {
     }
 
     protected SyncLog(SyncTypes.SyncType<?> syncType) {
-        this.syncDataType = syncType.ordinal();
-        this.clazz = syncType.dataType().getName();
+        this.syncType = syncType.ordinal();
+        // this.clazz = syncType.dataType().getName();
     }
 
     /** 日志序列号保证顺序 */
@@ -76,13 +76,13 @@ public class SyncLog implements Serializable {
     /**
      * {@link SyncTypes} ordinal
      */
-    protected int syncDataType;
+    protected int syncType;
 
     /** ================ 数据id标示:删除,更新用 ===================== */
     protected String dataKey;
 
     /** ================ 更新数据:类,更新方法,更新方法序列化后参数(避免修改) ===================== */
-    protected String clazz;
+    //protected String clazz;
 
     protected String method;
 
@@ -107,7 +107,7 @@ public class SyncLog implements Serializable {
 
     private byte[] toBytes0() {
         byte[] dataKeyBytes = dataKey == null ? new byte[0] : dataKey.getBytes();
-        byte[] clazzBytes = clazz == null ? new byte[0] : clazz.getBytes();
+        // byte[] clazzBytes = clazz == null ? new byte[0] : clazz.getBytes();
         byte[] methodBytes = method == null ? new byte[0] : method.getBytes();
         int argSize = args == null ? 0 : args.length;
 
@@ -145,7 +145,7 @@ public class SyncLog implements Serializable {
                 + 4 // dataKey 数据主键标识字节长度
                 + dataKeyBytes.length // dataKey 数据字节
                 + 4 // clazz 字节长度
-                + clazzBytes.length // clazz字节
+                // + clazzBytes.length // clazz字节
                 + 4 // method 字节长度
                 + methodBytes.length // method 字节
                 + 4 // args参数个数
@@ -160,7 +160,7 @@ public class SyncLog implements Serializable {
         Bytes.int2bytes(bytes.length - 6, bytes, offset);
         offset += 4;
 
-        bytes[offset] = (byte)syncDataType;
+        bytes[offset] = (byte) syncType;
         offset += 1;
         bytes[offset] = (byte)operateType;
         offset += 1;
@@ -173,10 +173,10 @@ public class SyncLog implements Serializable {
         System.arraycopy(dataKeyBytes, 0, bytes, offset, dataKeyBytes.length);
         offset += dataKeyBytes.length;
 
-        Bytes.int2bytes(clazzBytes.length, bytes, offset);
-        offset += 4;
-        System.arraycopy(clazzBytes, 0, bytes, offset, clazzBytes.length);
-        offset += clazzBytes.length;
+//        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;
@@ -220,16 +220,16 @@ public class SyncLog implements Serializable {
             buf.readBytes(dataKeyBytes);
             log.dataKey = new String(dataKeyBytes, StandardCharsets.UTF_8);
         }
-        int clazzLen = buf.readInt();
-        byte[] clazzBytes = new byte[clazzLen];
-        if (clazzLen > 0) {
-            buf.readBytes(clazzBytes);
-            log.clazz = new String(clazzBytes, StandardCharsets.UTF_8);
-        }
+//        int clazzLen = buf.readInt();
+//        byte[] clazzBytes = new byte[clazzLen];
+//        if (clazzLen > 0) {
+//            buf.readBytes(clazzBytes);
+//            log.clazz = new String(clazzBytes, StandardCharsets.UTF_8);
+//        }
         int methodLen = buf.readInt();
         if (methodLen > 0) {
             // 复用 clazz 字节数组
-            byte[] methodBytes = clazzLen >= methodLen ? clazzBytes : new byte[methodLen];
+            byte[] methodBytes = new byte[methodLen];
             buf.readBytes(methodBytes, 0, methodLen);
             log.method = new String(methodBytes, 0, methodLen, StandardCharsets.UTF_8);
         }

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

@@ -36,6 +36,7 @@ public class SyncCmdClientHandler extends SimpleChannelInboundHandler<SyncCmd> {
 
     private void handlePong(ChannelHandlerContext ctx, SyncCmd syncCmd) {
         logger.info("pong: {}", ctx.channel());
+        System.out.println("pong: " + ctx.channel());
     }
 
 }

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

@@ -3,6 +3,8 @@ package net.sopod.soim.router.datasync.server.handler;
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 import net.sopod.soim.router.datasync.server.data.SyncCmd;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * SyncCmdServerHandler
@@ -12,9 +14,29 @@ import net.sopod.soim.router.datasync.server.data.SyncCmd;
  */
 public class SyncCmdServerHandler extends SimpleChannelInboundHandler<SyncCmd> {
 
+    private static final Logger logger = LoggerFactory.getLogger(SyncCmdServerHandler.class);
+
     @Override
-    protected void channelRead0(ChannelHandlerContext channelHandlerContext, SyncCmd syncCmd) throws Exception {
+    protected void channelRead0(ChannelHandlerContext ctx, SyncCmd syncCmd) throws Exception {
+        switch (syncCmd.getCmdType()) {
+            case SyncCmd.PING:
+                this.handlePing(ctx, syncCmd);
+                break;
+            case SyncCmd.PONG:
+                this.handlePong(ctx, syncCmd);
+                break;
+        }
+    }
+
+    private void handlePing(ChannelHandlerContext ctx, SyncCmd syncCmd) {
+        logger.info("ping: {}", ctx.channel());
+        System.out.println("ping: " + ctx.channel());
+        SyncCmd pong = new SyncCmd().setCmdType(SyncCmd.PONG);
+        ctx.writeAndFlush(pong);
+    }
 
+    private void handlePong(ChannelHandlerContext ctx, SyncCmd syncCmd) {
+        logger.info("pong: {}", ctx.channel());
     }
 
 }

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

@@ -2,8 +2,13 @@ package net.sopod.soim.router.datasync.server.handler;
 
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
+import net.sopod.soim.router.datasync.DataSync;
+import net.sopod.soim.router.datasync.SyncTypes;
+import net.sopod.soim.router.datasync.server.codec.CodecUtil;
 import net.sopod.soim.router.datasync.server.data.SyncLog;
 
+import java.util.List;
+
 /**
  * SyncLogServerHandler
  * 即将删除的节点：会主动推送同步数据，这里进行接收
@@ -15,7 +20,13 @@ public class SyncLogServerHandler extends SimpleChannelInboundHandler<SyncLog> {
 
     @Override
     protected void channelRead0(ChannelHandlerContext channelHandlerContext, SyncLog syncLog) throws Exception {
-
+        List<byte[]> bytesList = syncLog.getSerializeDataCollect();
+        for (byte[] bytes : bytesList) {
+            SyncTypes.SyncType<DataSync> syncType = SyncTypes.getSyncType(syncLog.getSyncType());
+            DataSync instance = CodecUtil.decode(bytes, syncType.dataType());
+            System.out.println(instance);
+        }
+        System.out.println("read: " + syncLog);
     }
 
 }

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

@@ -0,0 +1,43 @@
+package net.sopod.soim.router.datasync.server.session;
+
+import net.sopod.soim.router.datasync.server.SyncClient;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * SyncClientSession
+ *
+ * @author tmy
+ * @date 2022-05-09 17:45
+ */
+public class SyncClientSession {
+
+    private static final SyncClientSession INSTANCE = new SyncClientSession();
+
+    public static SyncClientSession getInstance() {
+        return INSTANCE;
+    }
+
+    /**
+     * 服务端addr，连接 channel
+     */
+    public final ConcurrentHashMap<String, SyncClient> serverChannel = new ConcurrentHashMap<>();
+
+    public void connect(String host, int port) {
+        SyncClient client = new SyncClient();
+        try {
+            client.connect(host, port);
+            serverChannel.put(host + ":" + port, client);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public void close(String addr) {
+        SyncClient syncClient = serverChannel.get(addr);
+        if (syncClient != null) {
+            syncClient.close();
+        }
+    }
+
+}

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

@@ -0,0 +1,45 @@
+package net.sopod.soim.router.datasync.server.session;
+
+import org.apache.dubbo.common.utils.ConcurrentHashSet;
+
+import java.nio.channels.Channel;
+import java.util.Set;
+
+/**
+ * SyncServerSession
+ *
+ * @author tmy
+ * @date 2022-05-09 14:49
+ */
+public class SyncServerSession {
+
+    private static final SyncServerSession INSTANCE = new SyncServerSession();
+
+    public static SyncServerSession getInstance() {
+        return INSTANCE;
+    }
+
+    /** 已连接客户端 */
+    public final Set<Channel> clients = new ConcurrentHashSet<>();
+
+    /** 备份数据节点 */
+    public final Set<Channel> backupClients = new ConcurrentHashSet<>();
+
+    /** 新增节点发送 Sync 命令后添加到该连接集合 */
+    public final Set<Channel> newNodeClients = new ConcurrentHashSet<>();
+
+    public void addClient(Channel channel) {
+        clients.add(channel);
+    }
+
+    public void addBackupClient() {
+
+    }
+
+    public void removeClient(Channel channel) {
+        clients.remove(channel);
+        backupClients.remove(channel);
+        newNodeClients.remove(channel);
+    }
+
+}

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

@@ -8,6 +8,11 @@ import org.slf4j.LoggerFactory;
 /**
  * SyncSession
  * 作为客户端或服务端主动操作连接
+ * 服务端场景：
+ *   1.获取客户端连接(备份服务器类型，新增服务器同步类型)
+ * 客户端场景：
+ *   1.即将删除：推送到其他几个节点服务器
+ *   2.新增节点：连接几个节点服务器发送拉取数据命令
  *
  * @author tmy
  * @date 2022-05-08 17:34