From e4ba8102b96efd94db34b431169b544832d13196 Mon Sep 17 00:00:00 2001 From: tangmingyou <234767776@qq.com> Date: Mon, 9 May 2022 18:03:27 +0800 Subject: [PATCH] =?UTF-8?q?im-router=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E7=AB=AF=E5=AE=A2=E6=88=B7=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../router/config/ImRouterAppOnReady.java | 30 +++++++++++++ .../soim/router/datasync/SyncService.java | 5 +++ .../router/datasync/server/SyncClient.java | 30 +++++++++++++ .../router/datasync/server/data/SyncLog.java | 36 +++++++-------- .../server/handler/SyncCmdClientHandler.java | 1 + .../server/handler/SyncCmdServerHandler.java | 24 +++++++++- .../server/handler/SyncLogServerHandler.java | 13 +++++- .../server/session/SyncClientSession.java | 43 ++++++++++++++++++ .../server/session/SyncServerSession.java | 45 +++++++++++++++++++ .../datasync/server/session/SyncSession.java | 5 +++ 10 files changed, 212 insertions(+), 20 deletions(-) create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncClientSession.java create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncServerSession.java diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java b/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java index 9acf3ae..b83a151 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/config/ImRouterAppOnReady.java @@ -72,6 +72,36 @@ public class ImRouterAppOnReady implements ApplicationListener 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 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，使用了虚拟节点会导致迁移多个数据节点 * diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/SyncService.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/SyncService.java index cbab3d8..8613180 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/SyncService.java +++ b/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 routerUserMap = RouterUserStorage.getInstance().getRouterUserMap(); for (Map.Entry entry : routerUserMap.entrySet()) { diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncClient.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncClient.java index 20838be..ec22424 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncClient.java +++ b/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() { @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 addLog = SyncLog.addLog(1, SyncTypes.ROUTER_USER) + .addData(user) + .addData(user2); + client.clientChannel.writeAndFlush(addLog); + + Thread.sleep(10000); + client.close(); + } } diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/data/SyncLog.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/data/SyncLog.java index ed6f69b..7ff9e8b 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/data/SyncLog.java +++ b/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); } diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncCmdClientHandler.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncCmdClientHandler.java index dc1a4e5..fb97f87 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncCmdClientHandler.java +++ b/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 { private void handlePong(ChannelHandlerContext ctx, SyncCmd syncCmd) { logger.info("pong: {}", ctx.channel()); + System.out.println("pong: " + ctx.channel()); } } diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncCmdServerHandler.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncCmdServerHandler.java index 38d4396..171c837 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncCmdServerHandler.java +++ b/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 { + 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()); } } diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncLogServerHandler.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncLogServerHandler.java index ff02c64..808edf3 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/handler/SyncLogServerHandler.java +++ b/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 { @Override protected void channelRead0(ChannelHandlerContext channelHandlerContext, SyncLog syncLog) throws Exception { - + List bytesList = syncLog.getSerializeDataCollect(); + for (byte[] bytes : bytesList) { + SyncTypes.SyncType syncType = SyncTypes.getSyncType(syncLog.getSyncType()); + DataSync instance = CodecUtil.decode(bytes, syncType.dataType()); + System.out.println(instance); + } + System.out.println("read: " + syncLog); } } diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncClientSession.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncClientSession.java new file mode 100644 index 0000000..33a8160 --- /dev/null +++ b/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 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(); + } + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncServerSession.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncServerSession.java new file mode 100644 index 0000000..eb17871 --- /dev/null +++ b/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 clients = new ConcurrentHashSet<>(); + + /** 备份数据节点 */ + public final Set backupClients = new ConcurrentHashSet<>(); + + /** 新增节点发送 Sync 命令后添加到该连接集合 */ + public final Set 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); + } + +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncSession.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncSession.java index a65d28a..3e24c29 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/session/SyncSession.java +++ b/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