From 2c245e74c91e34be112bf7951cc49edc7df2835b Mon Sep 17 00:00:00 2001 From: tangmingyou <234767776@qq.com> Date: Thu, 5 May 2022 23:54:55 +0800 Subject: [PATCH] =?UTF-8?q?im-router=E6=95=B0=E6=8D=AE=E5=90=8C=E6=AD=A5?= =?UTF-8?q?=E6=97=A5=E5=BF=97?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../sopod/soim/router/cache/RouterUser.java | 1 + .../{server => }/DataChangeTrigger.java | 4 +- .../router/{cache => datasync}/DataSync.java | 2 +- .../DataSyncProxyFactory.java | 18 +- .../datasync/{server => }/SyncTypes.java | 10 +- .../{cache => datasync}/annotation/Sync.java | 2 +- .../annotation/SyncIgnore.java | 2 +- .../router/datasync/server/SyncDataIncr.java | 10 + .../soim/router/datasync/server/SyncLog.java | 297 ++++++++++++++---- 9 files changed, 274 insertions(+), 72 deletions(-) rename im-service/im-router/src/main/java/net/sopod/soim/router/datasync/{server => }/DataChangeTrigger.java (91%) rename im-service/im-router/src/main/java/net/sopod/soim/router/{cache => datasync}/DataSync.java (95%) rename im-service/im-router/src/main/java/net/sopod/soim/router/{cache => datasync}/DataSyncProxyFactory.java (92%) rename im-service/im-router/src/main/java/net/sopod/soim/router/datasync/{server => }/SyncTypes.java (93%) rename im-service/im-router/src/main/java/net/sopod/soim/router/{cache => datasync}/annotation/Sync.java (81%) rename im-service/im-router/src/main/java/net/sopod/soim/router/{cache => datasync}/annotation/SyncIgnore.java (82%) create mode 100644 im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncDataIncr.java diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/RouterUser.java b/im-service/im-router/src/main/java/net/sopod/soim/router/cache/RouterUser.java index 5aff674..19d5f9d 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/RouterUser.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/cache/RouterUser.java @@ -2,6 +2,7 @@ package net.sopod.soim.router.cache; import lombok.Data; import lombok.experimental.Accessors; +import net.sopod.soim.router.datasync.DataSync; /** * RouterUser diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/DataChangeTrigger.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataChangeTrigger.java similarity index 91% rename from im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/DataChangeTrigger.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataChangeTrigger.java index 7346e63..1029579 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/DataChangeTrigger.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataChangeTrigger.java @@ -1,6 +1,6 @@ -package net.sopod.soim.router.datasync.server; +package net.sopod.soim.router.datasync; -import net.sopod.soim.router.cache.DataSync; +import net.sopod.soim.router.datasync.server.SyncLog; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSync.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataSync.java similarity index 95% rename from im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSync.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataSync.java index 0e5a335..b6e9170 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSync.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataSync.java @@ -1,4 +1,4 @@ -package net.sopod.soim.router.cache; +package net.sopod.soim.router.datasync; /** * BiSync diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSyncProxyFactory.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataSyncProxyFactory.java similarity index 92% rename from im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSyncProxyFactory.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataSyncProxyFactory.java index dff73fc..05b098f 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/DataSyncProxyFactory.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/DataSyncProxyFactory.java @@ -1,24 +1,19 @@ -package net.sopod.soim.router.cache; +package net.sopod.soim.router.datasync; 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 net.sopod.soim.router.cache.RouterUser; +import net.sopod.soim.router.datasync.annotation.SyncIgnore; 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 @@ -34,7 +29,7 @@ public class DataSyncProxyFactory { /** * 缓存数据类型更新方法列表 */ - private static final Map, Set> cacheTypeUpdaterMethods = new ConcurrentHashMap<>(); + private static final Map, Set> typeUpdaterMethodsCache = new ConcurrentHashMap<>(); @SuppressWarnings("unchecked") public static T newProxyInstance(SyncTypes.SyncType syncType) { @@ -49,7 +44,7 @@ public class DataSyncProxyFactory { } // 获取查询更新方法列表 - Set updaterMethods = cacheTypeUpdaterMethods.computeIfAbsent(type, dataType -> { + Set updaterMethods = typeUpdaterMethodsCache.computeIfAbsent(type, dataType -> { Method[] methods = dataType.getMethods(); Set updaterMethodNames = new HashSet<>(); for (Method m : methods) { @@ -98,6 +93,7 @@ public class DataSyncProxyFactory { } @Override + @SuppressWarnings("unchecked") public Object intercept(Object instance, Method method, Object[] args, MethodProxy methodProxy) throws Throwable { String methodName = method.getName(); // 是判断不是更新的方法跳过 @@ -151,7 +147,7 @@ public class DataSyncProxyFactory { // return methodProxy.invokeSuper(o, args); // } // }); -// + // B b = (B) enhancer.create(); // b.setAge(12); // b.setName("沧海"); diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncTypes.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/SyncTypes.java similarity index 93% rename from im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncTypes.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/datasync/SyncTypes.java index 5629e37..189e3a5 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncTypes.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/SyncTypes.java @@ -1,7 +1,6 @@ -package net.sopod.soim.router.datasync.server; +package net.sopod.soim.router.datasync; 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; @@ -96,7 +95,12 @@ public class SyncTypes { @Nullable @SuppressWarnings("unchecked") static SyncType getSyncType(int ordinal) { - return (SyncType) (ordinal < TYPES.size() ? TYPES.get(ordinal) : null); + for (SyncType type : TYPES) { + if (type.ordinal == ordinal) { + return (SyncType) type; + } + } + return null; } @Override diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/Sync.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/annotation/Sync.java similarity index 81% rename from im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/Sync.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/datasync/annotation/Sync.java index 65c0f5e..e8268ac 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/Sync.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/annotation/Sync.java @@ -1,4 +1,4 @@ -package net.sopod.soim.router.cache.annotation; +package net.sopod.soim.router.datasync.annotation; import java.lang.annotation.*; diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/SyncIgnore.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/annotation/SyncIgnore.java similarity index 82% rename from im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/SyncIgnore.java rename to im-service/im-router/src/main/java/net/sopod/soim/router/datasync/annotation/SyncIgnore.java index 26333cd..9e20a73 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/cache/annotation/SyncIgnore.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/annotation/SyncIgnore.java @@ -1,4 +1,4 @@ -package net.sopod.soim.router.cache.annotation; +package net.sopod.soim.router.datasync.annotation; import java.lang.annotation.*; diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncDataIncr.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncDataIncr.java new file mode 100644 index 0000000..e92a6b9 --- /dev/null +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncDataIncr.java @@ -0,0 +1,10 @@ +package net.sopod.soim.router.datasync.server; + +/** + * SyncDataIncr + * + * @author tmy + * @date 2022-05-05 22:33 + */ +public class SyncDataIncr { +} diff --git a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncLog.java b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncLog.java index a411674..5924443 100644 --- a/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncLog.java +++ b/im-service/im-router/src/main/java/net/sopod/soim/router/datasync/server/SyncLog.java @@ -1,9 +1,14 @@ package net.sopod.soim.router.datasync.server; import io.netty.buffer.ByteBuf; -import lombok.Data; +import io.netty.buffer.Unpooled; +import lombok.Getter; +import lombok.ToString; import lombok.experimental.Accessors; import net.sopod.soim.common.util.Jackson; +import net.sopod.soim.router.cache.RouterUser; +import net.sopod.soim.router.datasync.DataSync; +import net.sopod.soim.router.datasync.SyncTypes; import org.apache.dubbo.common.io.Bytes; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -12,15 +17,19 @@ import javax.annotation.Nullable; import java.io.Serializable; import java.lang.reflect.Method; import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; /** - * SyncLog - * syncLog serialize + * SyncReviseLog + * 数据修改日志（增删改） * * @author tmy * @date 2022-05-05 11:37 */ -@Data +@Getter +@ToString @Accessors(chain = true) public class SyncLog implements Serializable { @@ -34,8 +43,25 @@ public class SyncLog implements Serializable { public static final int OPT_REMOVE = 2; public static final int OPT_UPDATE = 3; + public static AddLog addLog(SyncTypes.SyncType syncType) { + return new AddLog<>(syncType); + } + + public static UpdateLog updateLog(int logSeq, SyncTypes.SyncType syncType) { + return new UpdateLog<>(logSeq, syncType); + } + + public static RemoveLog removeLog(int logSeq, SyncTypes.SyncType syncType) { + return new RemoveLog<>(logSeq, syncType); + } + + protected SyncLog(SyncTypes.SyncType syncType) { + this.syncDataType = syncType.ordinal(); + this.clazz = syncType.dataType().getName(); + } + /** 日志序列号保证顺序 */ - private int logSeq; + protected int logSeq; /** * 操作类型: @@ -43,56 +69,86 @@ public class SyncLog implements Serializable { * 2.删除 * 3.更新 */ - private int operateType; + protected int operateType; /** * {@link SyncTypes} ordinal */ - private int syncDataType; + protected int syncDataType; + + /** ================ 数据id标示:删除,更新用 ===================== */ + protected String dataKey; - /** 数据id标示 */ - private String dataKey; + /** ================ 更新数据:类,更新方法,更新方法序列化参数(避免修改) ===================== */ + protected String clazz; - /** ================ 新增参数:序列化后的数据 ===================== */ - private String addSerializeData; + protected String method; - /** ================ 更新参数 ===================== */ - private String clazz; + protected String[] args; - private String method; + /** ================ 新增数据:序列化后的数据(避免修改) ===================== */ + protected List serializeDataCollect; - private Object[] args; + /** 数据可能同步给多个订阅者，缓存一下 */ + private transient byte[] toBytesCache; public byte[] toBytes() { - byte[] clazzBytes = clazz.getBytes(); - byte[] methodBytes = method.getBytes(); + if (toBytesCache == null) { + synchronized (this) { + if (toBytesCache == null) { + toBytesCache = toBytes0(); + } + } + } + return toBytesCache; + } + + private byte[] toBytes0() { + byte[] dataKeyBytes = dataKey == null ? new byte[0] : dataKey.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; byte[][] byteArgs = new byte[argSize][]; - if (args != null) { + int argsByteLen = 0; + if (argSize > 0) { 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); + byteArgs[i] = args[i].getBytes(StandardCharsets.UTF_8); + argsByteLen += 4; + argsByteLen += byteArgs[i].length; } } - int argsByteLen = 0; - for (byte[] byteArg : byteArgs) { - argsByteLen += 4; - argsByteLen += byteArg.length; + + int dataCollectSize = serializeDataCollect == null ? 0 : serializeDataCollect.size(); + byte[][] byteData = new byte[dataCollectSize][]; + int dataCollectByteLen = 0; + if (dataCollectSize > 0) { + int i = 0; + for (String dataCollect : serializeDataCollect) { + byteData[i] = dataCollect.getBytes(StandardCharsets.UTF_8); + dataCollectByteLen += 4; + dataCollectByteLen += byteData[i].length; + i++; + } } // 总长 + 同步数据类型(byte) + byte[] bytes = new byte[ 2 // 魔术 + 4 // 请求体总长度 - + 1 // 同步数据类型 + + 1 // 同步数据类型 SyncType + + 1 // 新增/删除/更新 + + 4 // logSeq 序列号 + + 4 // dataKey 数据主键标识字节长度 + + dataKeyBytes.length // dataKey 数据字节 + 4 // clazz 字节长度 + clazzBytes.length // clazz字节 + 4 // method 字节长度 + methodBytes.length // method 字节 + 4 // args参数个数 - + argsByteLen // args参数字节 + + argsByteLen // args参数字节(len,argByte,len,argByte...) + + 4 // 序列化数据个数 + + dataCollectByteLen // 序列化数据字节(len,dataBytes,len,dataBytes...) ]; int offset = 0; Bytes.short2bytes(MAGIC, bytes, offset); @@ -103,6 +159,16 @@ public class SyncLog implements Serializable { bytes[offset] = (byte)syncDataType; offset += 1; + bytes[offset] = (byte)operateType; + offset += 1; + + Bytes.int2bytes(logSeq, bytes, offset); + offset += 4; + + Bytes.int2bytes(dataKeyBytes.length, bytes, offset); + offset += 4; + System.arraycopy(dataKeyBytes, 0, bytes, offset, dataKeyBytes.length); + offset += dataKeyBytes.length; Bytes.int2bytes(clazzBytes.length, bytes, offset); offset += 4; @@ -116,7 +182,7 @@ public class SyncLog implements Serializable { Bytes.int2bytes(argSize, bytes, offset); offset += 4; - if (argsByteLen > 0) { + if (argSize > 0) { for (byte[] byteArg : byteArgs) { Bytes.int2bytes(byteArg.length, bytes, offset); offset += 4; @@ -124,6 +190,16 @@ public class SyncLog implements Serializable { offset += byteArg.length; } } + Bytes.int2bytes(dataCollectSize, bytes, offset); + offset += 4; + if (dataCollectSize > 0) { + for (byte[] biData : byteData) { + Bytes.int2bytes(biData.length, bytes, offset); + offset += 4; + System.arraycopy(biData, 0, bytes, offset, biData.length); + offset += biData.length; + } + } return bytes; } @@ -132,46 +208,99 @@ public class SyncLog implements Serializable { if (magic != MAGIC) { throw new IllegalStateException("unknown bytes magic error"); } - SyncLog log = new SyncLog(); // 后续bytes长度 int dataLen = buf.readInt(); - log.syncDataType = buf.readByte(); + int syncDataType = buf.readByte(); + SyncTypes.SyncType syncType = SyncTypes.getSyncType(syncDataType); + SyncLog log = new SyncLog(syncType); + log.operateType = buf.readByte(); + log.logSeq = buf.readInt(); + int dataKeyLen = buf.readInt(); + if (dataKeyLen > 0) { + byte[] dataKeyBytes = new byte[dataKeyLen]; + buf.readBytes(dataKeyBytes); + log.dataKey = new String(dataKeyBytes, StandardCharsets.UTF_8); + } int clazzLen = buf.readInt(); byte[] clazzBytes = new byte[clazzLen]; - buf.readBytes(clazzBytes); - log.clazz = new String(clazzBytes, StandardCharsets.UTF_8); + if (clazzLen > 0) { + 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); + if (methodLen > 0) { + // 复用 clazz 字节数组 + 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]; + log.args = new String[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 + "个"); - } +// 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; + log.args[i] = new String(argBytes, 0, argLen, StandardCharsets.UTF_8); + } + } + int dataSize = buf.readInt(); + log.serializeDataCollect = dataSize > 0 ? new ArrayList<>(dataSize) : Collections.emptyList(); + if (dataSize > 0) { + byte[] dataByte = new byte[0]; + for (int i = 0; i < dataSize; i++) { + int dataByteLen = buf.readInt(); + dataByte = dataByte.length >= dataByteLen ? dataByte : new byte[dataByteLen]; + buf.readBytes(dataByte, 0, dataByteLen); + log.serializeDataCollect.add(new String(dataByte, 0, dataByteLen, StandardCharsets.UTF_8)); } } return log; } + public static void main(String[] args) { + RouterUser user1 = new RouterUser() + .setUid(10001L) + .setAccount("前线") + .setImEntryAddr("127.0.0.1") + .setIsOnline(false); + RouterUser user2 = new RouterUser() + .setUid(10002L) + .setAccount("画中") + .setImEntryAddr("127.0.0.2") + .setIsOnline(true); + AddLog addLog = addLog(SyncTypes.ROUTER_USER) + .addSerializeData(user1) + .addSerializeData(user2); +// String json = Jackson.json().serialize(addLog); +// System.out.println(json); +// System.out.println(json.getBytes(StandardCharsets.UTF_8).length); + + byte[] bytes = addLog.toBytes(); + ByteBuf buf = Unpooled.wrappedBuffer(bytes); + SyncLog newLog = SyncLog.read(buf); + buf.release(); + System.out.println(bytes.length); + System.out.println(addLog); + System.out.println(newLog); + } + + /** * TODO 缓存反射结果 */ + @Deprecated @Nullable private static Method getClassMethod(String clazzName, String methodName) { try { @@ -188,13 +317,75 @@ public class SyncLog implements Serializable { return null; } - public static void main(String[] args) { - new SyncLog() - .setSyncDataType(SyncTypes.ROUTER_USER.ordinal()) - .setClazz("") - .setMethod("") - .setArgs(new Object[]{}); + /** + * 新增数据 + * @param + */ + public static class AddLog extends SyncLog { + AddLog(SyncTypes.SyncType syncType) { + super(syncType); + this.operateType = OPT_ADD; + } + public AddLog setSerializeDataCollect(List serializeDataCollect) { + this.serializeDataCollect = serializeDataCollect; + return this; + } + + public AddLog addSerializeData(T data) { + if (this.serializeDataCollect == null) { + this.serializeDataCollect = new ArrayList<>(); + } + this.serializeDataCollect.add(Jackson.json().serialize(data)); + return this; + } + } + + /** + * 更新数据 + * @param + */ + public static class UpdateLog extends SyncLog { + UpdateLog(int logSeq, SyncTypes.SyncType syncType) { + super(syncType); + this.operateType = OPT_UPDATE; + this.logSeq = logSeq; + } + + public UpdateLog setDataKey(String dataKey) { + this.dataKey = dataKey; + return this; + } + + public UpdateLog setMethod(String method) { + this.method = method; + return this; + } + + public UpdateLog setArgs(Object[] args) { + this.args = new String[args.length]; + for (int i = 0; i < args.length; i++) { + this.args[i] = Jackson.json().serialize(args[i]); + } + return this; + } + } + + /** + * 删除数据 + * @param + */ + public static class RemoveLog extends SyncLog { + RemoveLog(int logSeq, SyncTypes.SyncType syncType) { + super(syncType); + this.operateType = OPT_REMOVE; + this.logSeq = logSeq; + } + + public RemoveLog setDataKey(String dataKey) { + this.dataKey = dataKey; + return this; + } } }