From 5f14eeae831e26661381f99a32d99ecc087ec48d Mon Sep 17 00:00:00 2001
From: godotg <jaysunxiao@gmail.com>
Date: Tue, 25 Jun 2024 15:05:08 +0800
Subject: [PATCH] perf[orm]: thread safe entity

---
 .../java/com/zfoo/orm/cache/EntityCache.java  | 39 +++++++++----------
 .../java/com/zfoo/orm/manager/OrmManager.java |  2 +-
 .../java/com/zfoo/orm/model/EntityDef.java    | 11 +++---
 3 files changed, 26 insertions(+), 26 deletions(-)

diff --git a/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java b/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java
index 30b527cb..2b51e176 100644
--- a/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java
+++ b/orm/src/main/java/com/zfoo/orm/cache/EntityCache.java
@@ -46,8 +46,8 @@ public class EntityCache<PK extends Comparable<PK>, E extends IEntity<PK>> imple
 
     private static final Logger logger = LoggerFactory.getLogger(EntityCache.class);
 
-    private static final int BATCH_SIZE = 512;
-    private static final int UNSAFE_COLLECTION_BATCH_SIZE = BATCH_SIZE / Runtime.getRuntime().availableProcessors();
+    private static final int DEFAULT_BATCH_SIZE = 512;
+    private static final int NOT_THREAD_SAFE_BATCH_SIZE = Math.max(128, DEFAULT_BATCH_SIZE / Runtime.getRuntime().availableProcessors());
 
     private final EntityDef entityDef;
 
@@ -237,7 +237,7 @@ public class EntityCache<PK extends Comparable<PK>, E extends IEntity<PK>> imple
         pnode.resetTime(TimeUtils.currentTimeMillis());
         var updateList = new ArrayList<E>();
         updateList.add(pnode.getEntity());
-        doPersist(updateList, BATCH_SIZE);
+        doPersist(updateList, DEFAULT_BATCH_SIZE);
     }
 
     // 游戏中80%都是执行更新的操作，这样做会极大的提高更新速度
@@ -246,8 +246,20 @@ public class EntityCache<PK extends Comparable<PK>, E extends IEntity<PK>> imple
     @Override
     public void persistAll() {
         var currentTime = TimeUtils.currentTimeMillis();
-
-        if (entityDef.hasUnsafeCollection()) {
+        if (entityDef.isThreadSafe()) {
+            var updateList = new ArrayList<E>();
+            cache.forEach(new BiConsumer<PK, PNode<E>>() {
+                @Override
+                public void accept(PK pk, PNode<E> pnode) {
+                    var entity = pnode.getEntity();
+                    if (pnode.getModifiedTime() != pnode.getWriteToDbTime()) {
+                        pnode.resetTime(currentTime);
+                        updateList.add(entity);
+                    }
+                }
+            });
+            EventBus.asyncExecute(entityDef.getClazz().hashCode(), () -> doPersist(updateList, DEFAULT_BATCH_SIZE));
+        } else {
             // key为threadId
             var updateMap = new HashMap<Long, List<E>>();
             cache.forEach(new BiConsumer<PK, PNode<E>>() {
@@ -266,24 +278,11 @@ public class EntityCache<PK extends Comparable<PK>, E extends IEntity<PK>> imple
                 var updateList = entry.getValue();
                 var executor = ThreadUtils.executorByThreadId(threadId);
                 if (executor == null) {
-                    EventBus.asyncExecute(entityDef.getClazz().hashCode(), () -> doPersist(updateList, BATCH_SIZE));
+                    EventBus.asyncExecute(entityDef.getClazz().hashCode(), () -> doPersist(updateList, DEFAULT_BATCH_SIZE));
                 } else {
-                    executor.execute(() -> doPersist(updateList, UNSAFE_COLLECTION_BATCH_SIZE));
+                    executor.execute(() -> doPersist(updateList, NOT_THREAD_SAFE_BATCH_SIZE));
                 }
             }
-        } else {
-            var updateList = new ArrayList<E>();
-            cache.forEach(new BiConsumer<PK, PNode<E>>() {
-                @Override
-                public void accept(PK pk, PNode<E> pnode) {
-                    var entity = pnode.getEntity();
-                    if (pnode.getModifiedTime() != pnode.getWriteToDbTime()) {
-                        pnode.resetTime(currentTime);
-                        updateList.add(entity);
-                    }
-                }
-            });
-            EventBus.asyncExecute(entityDef.getClazz().hashCode(), () -> doPersist(updateList, BATCH_SIZE));
         }
     }
 
diff --git a/orm/src/main/java/com/zfoo/orm/manager/OrmManager.java b/orm/src/main/java/com/zfoo/orm/manager/OrmManager.java
index 96abdd75..70663378 100644
--- a/orm/src/main/java/com/zfoo/orm/manager/OrmManager.java
+++ b/orm/src/main/java/com/zfoo/orm/manager/OrmManager.java
@@ -376,7 +376,7 @@ public class OrmManager implements IOrmManager {
             indexTextDefMap.put(field.getName(), indexTextDef);
         }
 
-        return EntityDef.valueOf(idField, clazz, hasUnsafeCollection, cacheSize, expireMillisecond, persisterStrategy, indexDefMap, indexTextDefMap);
+        return EntityDef.valueOf(idField, clazz, !hasUnsafeCollection, cacheSize, expireMillisecond, persisterStrategy, indexDefMap, indexTextDefMap);
     }
 
     private void checkEntity(Class<?> clazz) {
diff --git a/orm/src/main/java/com/zfoo/orm/model/EntityDef.java b/orm/src/main/java/com/zfoo/orm/model/EntityDef.java
index d986f524..6a0fbbe7 100644
--- a/orm/src/main/java/com/zfoo/orm/model/EntityDef.java
+++ b/orm/src/main/java/com/zfoo/orm/model/EntityDef.java
@@ -28,7 +28,8 @@ public class EntityDef {
 
     private Class<? extends IEntity<?>> clazz;
 
-    private boolean hasUnsafeCollection;
+    // 线程安全指的是内部没有使用集合或者使用的集合全部支持并发操作
+    private boolean threadSafe;
 
     private int cacheSize;
 
@@ -41,12 +42,12 @@ public class EntityDef {
     private Map<String, IndexTextDef> indexTextDefMap;
 
 
-    public static EntityDef valueOf(Field idField, Class<? extends IEntity<?>> clazz, boolean hasUnsafeCollection, int cacheSize, long expireMillisecond
+    public static EntityDef valueOf(Field idField, Class<? extends IEntity<?>> clazz, boolean threadSafe, int cacheSize, long expireMillisecond
             , PersisterStrategy persisterStrategy, Map<String, IndexDef> indexDefMap, Map<String, IndexTextDef> indexTextDefMap) {
         var entityDef = new EntityDef();
         entityDef.idField = idField;
         entityDef.clazz = clazz;
-        entityDef.hasUnsafeCollection = hasUnsafeCollection;
+        entityDef.threadSafe = threadSafe;
         entityDef.cacheSize = cacheSize;
         entityDef.expireMillisecond = expireMillisecond;
         entityDef.persisterStrategy = persisterStrategy;
@@ -71,8 +72,8 @@ public class EntityDef {
         return clazz;
     }
 
-    public boolean hasUnsafeCollection() {
-        return hasUnsafeCollection;
+    public boolean isThreadSafe() {
+        return threadSafe;
     }
 
     public int getCacheSize() {