diff --git a/common/src/main/java/com/ljsd/common/mogodb/LjsdMongoTemplate.java b/common/src/main/java/com/ljsd/common/mogodb/LjsdMongoTemplate.java index 5eb4f4cd2..eed4c9606 100644 --- a/common/src/main/java/com/ljsd/common/mogodb/LjsdMongoTemplate.java +++ b/common/src/main/java/com/ljsd/common/mogodb/LjsdMongoTemplate.java @@ -9,6 +9,7 @@ import java.lang.reflect.Field; import java.util.ArrayList; import java.util.Iterator; import java.util.List; +import java.util.Map; public class LjsdMongoTemplate { @@ -78,6 +79,10 @@ public class LjsdMongoTemplate { MongoUpdateCacheThreadLocal.update(this); } + public void concurrentDataUpdate(Map> updateRequestMap) throws Exception { + MongoUpdateCacheThreadLocal.concurrentDataUpdate(this,updateRequestMap); + } + public int inc(String collectionName, String id, String key){ DBCollection collection = mongoTemplate.getCollection(collectionName); BasicDBObject searchQuery = new BasicDBObject(); diff --git a/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCache.java b/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCache.java index ef9dbaef1..6fc5f4a05 100644 --- a/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCache.java +++ b/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCache.java @@ -1,13 +1,17 @@ package com.ljsd.common.mogodb; import com.ljsd.common.mogodb.util.BlockingUniqueQueue; +import com.ljsd.common.mogodb.util.ConcurrentDataMessage; +import com.ljsd.common.mogodb.util.MongoKeys; import com.mongodb.BasicDBObject; import java.util.*; import java.util.concurrent.BlockingQueue; +import java.util.concurrent.ConcurrentHashMap; public class MongoUpdateCache { - static class UpdateRequest { + + public static class UpdateRequest { MongoBase mongoBase; String fullKey; Object value; @@ -23,38 +27,42 @@ public class MongoUpdateCache { //key:String, collection + ":" + id private final Map> requestMap = new HashMap<>(); - static final BlockingQueue>> logQueue = new BlockingUniqueQueue<>(); + private final Map> updateDataMap = new HashMap<>(); protected void addUpdateRequest(MongoBase mongoBase, String fullKey, Object value, int opCode) { - String key = mongoBase.getRootCollection() + ":" + mongoBase.getRootId(); - Map listRequest = requestMap.get(key); - if (listRequest == null) { - listRequest = new HashMap<>(); - requestMap.put(key, listRequest); - } else { - Iterator> it = listRequest.entrySet().iterator(); - while (it.hasNext()) { - Map.Entry entry = it.next(); - String key1 = entry.getKey(); - if(!fullKey.equals(key1)){ - //解决类似 item.1 item.11 问题 - if (fullKey.contains(key1)) { - if(checkOverride(key1,fullKey)){ - return; - } - continue; + String key = mongoBase.getRootCollection() + ":" + mongoBase.getRootId(); + Map> dataMap = requestMap; + if (MongoKeys.mongoKeyMap.containsKey(fullKey.split("\\.")[0]) || MongoKeys.mongoKeyMap.containsKey(fullKey)) { + dataMap = updateDataMap; + } + Map listRequest = dataMap.get(key); + if (listRequest == null) { + listRequest = new HashMap<>(); + dataMap.put(key, listRequest); + } else { + Iterator> it = listRequest.entrySet().iterator(); + while (it.hasNext()) { + Map.Entry entry = it.next(); + String key1 = entry.getKey(); + if (!fullKey.equals(key1)) { + //解决类似 item.1 item.11 问题 + if (fullKey.contains(key1)) { + if (checkOverride(key1, fullKey)) { + return; } - if (key1.contains(fullKey)) { - if(checkOverride(key1,fullKey)){ - it.remove(); - } + continue; + } + if (key1.contains(fullKey)) { + if (checkOverride(key1, fullKey)) { + it.remove(); } } - } + } - UpdateRequest request = new UpdateRequest(fullKey, value, opCode, mongoBase); - listRequest.put(fullKey, request); + } + UpdateRequest request = new UpdateRequest(fullKey, value, opCode, mongoBase); + listRequest.put(fullKey, request); } public boolean checkOverride(String key1,String key2){ @@ -67,15 +75,25 @@ public class MongoUpdateCache { } + protected void update(LjsdMongoTemplate ljsdMongoTemplate) { + updateMongoData(ljsdMongoTemplate, requestMap); + requestMap.clear(); + if (updateDataMap.size() !=0){ + ConcurrentDataMessage.addLogQueue(updateDataMap); + updateDataMap.clear(); + } + } - protected void update(LjsdMongoTemplate ljsdMongoTemplate) throws IllegalAccessException { + public void concurrentDataUpdate(LjsdMongoTemplate ljsdMongoTemplate, Map> updateRequestMap) { + updateMongoData(ljsdMongoTemplate, updateRequestMap); + } + + private void updateMongoData(LjsdMongoTemplate ljsdMongoTemplate, Map> updateRequestMap) { BasicDBObject searchQuery = new BasicDBObject(); - - for (Map.Entry> entry : requestMap.entrySet()) { + for (Map.Entry> entry : updateRequestMap.entrySet()) { String[] strs = entry.getKey().split(":"); String collection = strs[0]; String id = strs[1]; - searchQuery.put("_id", Integer.parseInt(id)); BasicDBObject valueDBObj = new BasicDBObject(); BasicDBObject dbobj = new BasicDBObject(); @@ -98,12 +116,10 @@ public class MongoUpdateCache { dbobj.append("$unset", removedbobj); flag = true; } - if (flag){ + if (flag) { ljsdMongoTemplate.updateValue(collection, searchQuery, dbobj); } - } - requestMap.clear(); } } diff --git a/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCacheThreadLocal.java b/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCacheThreadLocal.java index 2f2aeb3d6..ccabaf6a4 100644 --- a/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCacheThreadLocal.java +++ b/common/src/main/java/com/ljsd/common/mogodb/MongoUpdateCacheThreadLocal.java @@ -1,12 +1,11 @@ package com.ljsd.common.mogodb; -import java.util.HashMap; import java.util.Map; public class MongoUpdateCacheThreadLocal { private final static ThreadLocal mongoUpdateCacheThreadLocal = new ThreadLocal<>(); - private static MongoUpdateCache getMongoUpdateCache() { + public static MongoUpdateCache getMongoUpdateCache() { MongoUpdateCache mongoUpdateCache = mongoUpdateCacheThreadLocal.get(); if (mongoUpdateCache == null) { mongoUpdateCache = new MongoUpdateCache(); @@ -29,4 +28,9 @@ public class MongoUpdateCacheThreadLocal { MongoUpdateCache mongoUpdateCache = getMongoUpdateCache(); mongoUpdateCache.update(ljsdMongoTemplate); } + + public static void concurrentDataUpdate(LjsdMongoTemplate ljsdMongoTemplate,Map> updateRequestMap) { + MongoUpdateCache mongoUpdateCache = getMongoUpdateCache(); + mongoUpdateCache.concurrentDataUpdate(ljsdMongoTemplate, updateRequestMap); + } } diff --git a/common/src/main/java/com/ljsd/common/mogodb/util/ConcurrentDataMessage.java b/common/src/main/java/com/ljsd/common/mogodb/util/ConcurrentDataMessage.java new file mode 100644 index 000000000..e943c164d --- /dev/null +++ b/common/src/main/java/com/ljsd/common/mogodb/util/ConcurrentDataMessage.java @@ -0,0 +1,21 @@ +package com.ljsd.common.mogodb.util; + +import com.ljsd.common.mogodb.LjsdMongoTemplate; +import com.ljsd.common.mogodb.MongoUpdateCache; + +import java.util.Map; +import java.util.concurrent.BlockingQueue; + +public class ConcurrentDataMessage { +// public static final BlockingQueue dataQueue = new BlockingUniqueQueue<>(); + static BlockingQueue>> dataQueue = new BlockingUniqueQueue<>(); + //入队列 + public static void addLogQueue(Map> info){ + dataQueue.offer(info); + } + + public static void manageData(LjsdMongoTemplate myMongoTemplate) throws Exception { + Map> take = ConcurrentDataMessage.dataQueue.take(); + myMongoTemplate.concurrentDataUpdate(take); + } +} diff --git a/common/src/main/java/com/ljsd/common/mogodb/util/MongoKeys.java b/common/src/main/java/com/ljsd/common/mogodb/util/MongoKeys.java new file mode 100644 index 000000000..99e6a7b2e --- /dev/null +++ b/common/src/main/java/com/ljsd/common/mogodb/util/MongoKeys.java @@ -0,0 +1,13 @@ +package com.ljsd.common.mogodb.util; + +import java.util.HashMap; +import java.util.Map; + +public class MongoKeys { + + public static Map mongoKeyMap = new HashMap<>(); + + public static void initmongoKey() { + mongoKeyMap.put("friendManager",1); + } +} diff --git a/serverlogic/src/main/java/com/ljsd/GameApplication.java b/serverlogic/src/main/java/com/ljsd/GameApplication.java index e67bee369..1be4d7d3f 100644 --- a/serverlogic/src/main/java/com/ljsd/GameApplication.java +++ b/serverlogic/src/main/java/com/ljsd/GameApplication.java @@ -1,5 +1,6 @@ package com.ljsd; +import com.ljsd.common.mogodb.util.MongoKeys; import com.ljsd.jieling.config.ServerConfig; import com.ljsd.jieling.config.json.ServerConfiguration; import com.ljsd.jieling.config.json.ServerProperties; @@ -19,6 +20,7 @@ import com.ljsd.jieling.network.NettyServerAutoConfiguration; import com.ljsd.jieling.network.server.ProtocolsManager; import com.ljsd.jieling.network.session.ISessionFactory; import com.ljsd.jieling.thread.ThreadManager; +import com.ljsd.jieling.thread.task.MongoDataHandlerTask; import com.ljsd.jieling.util.KeyGenUtils; import com.ljsd.jieling.util.SensitivewordFilter; import com.ljsd.jieling.util.TimeUtils; @@ -82,14 +84,16 @@ public class GameApplication { // HttpPool.init(); STableManager.initialize("com.ljsd.jieling.config"); + //mongo异步数据处理线程 + MongoDataHandlerTask.init(configurableApplicationContext); //初始化邮件 MailingSystemManager.init(configurableApplicationContext); ThreadManager threadManager = ThreadManager.getInstance(); threadManager.init(configurableApplicationContext); MapLogic.getInstance().init(configurableApplicationContext); - + MongoKeys.initmongoKey(); SensitivewordFilter.init(); - HttpPool.init(); + HttpPool.init(); // 加载lua文件 CheckFight.getInstance().init("luafight/BattleMain.lua"); diff --git a/serverlogic/src/main/java/com/ljsd/jieling/config/reportData/DataMessageUtils.java b/serverlogic/src/main/java/com/ljsd/jieling/config/reportData/DataMessageUtils.java index 2180727d4..db9bbdad7 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/config/reportData/DataMessageUtils.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/config/reportData/DataMessageUtils.java @@ -9,7 +9,6 @@ import com.ljsd.jieling.util.http.HttpPool; import java.util.ArrayList; import java.util.List; import java.util.concurrent.BlockingQueue; -import java.util.concurrent.CopyOnWriteArrayList; public class DataMessageUtils { diff --git a/serverlogic/src/main/java/com/ljsd/jieling/thread/ThreadManager.java b/serverlogic/src/main/java/com/ljsd/jieling/thread/ThreadManager.java index d8c02cb7c..b01c3ddc9 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/thread/ThreadManager.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/thread/ThreadManager.java @@ -1,10 +1,7 @@ package com.ljsd.jieling.thread; -import com.ljsd.jieling.thread.task.DataReportTask; -import com.ljsd.jieling.thread.task.MinuteTask; -import com.ljsd.jieling.thread.task.PlatConfigureTask; -import com.ljsd.jieling.thread.task.RetrySendIndicationThread; +import com.ljsd.jieling.thread.task.*; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.context.ConfigurableApplicationContext; @@ -25,6 +22,7 @@ public class ThreadManager { private static PlatConfigureTask platConfigureTask; private static DataReportTask dataReportTask; + private static MongoDataHandlerTask mongoDataHandlerTask; public static int SLEEP_INTEVAL_TIME = 60; //每1分钟检查一次 @@ -37,8 +35,10 @@ public class ThreadManager { public void init(ConfigurableApplicationContext configurableApplicationContext) { platConfigureTask = configurableApplicationContext.getBean(PlatConfigureTask.class); - DataReportTask dataReportTask = new DataReportTask(); - dataReportTask.start(); + mongoDataHandlerTask = configurableApplicationContext.getBean(MongoDataHandlerTask.class); + dataReportTask = configurableApplicationContext.getBean(DataReportTask.class); +// DataReportTask dataReportTask = new DataReportTask(); +// dataReportTask.start(); go(); } @@ -84,6 +84,8 @@ public class ThreadManager { int delayForMinute = 60 - (now%60); LOGGER.info("delayForMinute---->>>>{}" ,delayForMinute); scheduledExecutor.scheduleAtFixedRate(platConfigureTask, 10, SLEEP_INTEVAL_TIME, TimeUnit.SECONDS); + scheduledExecutor.scheduleAtFixedRate(dataReportTask, 10, SLEEP_INTEVAL_TIME, TimeUnit.SECONDS); + scheduledExecutor.scheduleAtFixedRate(mongoDataHandlerTask, 10, SLEEP_INTEVAL_TIME, TimeUnit.SECONDS); scheduledExecutor.scheduleAtFixedRate(new MinuteTask(), delayForMinute, 60, TimeUnit.SECONDS); scheduledExecutor.scheduleAtFixedRate(new Thread(){ public void run () { diff --git a/serverlogic/src/main/java/com/ljsd/jieling/thread/task/DataReportTask.java b/serverlogic/src/main/java/com/ljsd/jieling/thread/task/DataReportTask.java index 2b26183ff..548e38643 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/thread/task/DataReportTask.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/thread/task/DataReportTask.java @@ -5,10 +5,11 @@ import org.springframework.stereotype.Component; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +@Component public class DataReportTask extends Thread { public DataReportTask(){ - setName("DataReportTaskThread"); + setName("DataReportTask"); } @Override public void run() { diff --git a/serverlogic/src/main/java/com/ljsd/jieling/thread/task/MongoDataHandlerTask.java b/serverlogic/src/main/java/com/ljsd/jieling/thread/task/MongoDataHandlerTask.java new file mode 100644 index 000000000..ed32db6c2 --- /dev/null +++ b/serverlogic/src/main/java/com/ljsd/jieling/thread/task/MongoDataHandlerTask.java @@ -0,0 +1,38 @@ +package com.ljsd.jieling.thread.task; + +import com.ljsd.common.mogodb.LjsdMongoTemplate; +import com.ljsd.common.mogodb.util.ConcurrentDataMessage; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.stereotype.Component; + +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +@Component +public class MongoDataHandlerTask extends Thread { + private static LjsdMongoTemplate myMongoTemplate; + public MongoDataHandlerTask(){ + setName("MongoDataHandlerTask"); + } + @Override + public void run() { + while (true) { + try { + ConcurrentDataMessage.manageData(myMongoTemplate); + } catch (Exception e) { + e.printStackTrace(); + } + } + } + public static void init(ConfigurableApplicationContext configurableApplicationContext){ + myMongoTemplate =configurableApplicationContext.getBean(LjsdMongoTemplate.class); +// startTlogThread(); + } + + public static void startTlogThread(){ + ExecutorService pool = Executors.newCachedThreadPool(); + for (int i = 0; i < 1 ; i++){ + MongoDataHandlerTask myThread = new MongoDataHandlerTask(); + pool.execute(myThread); + } + } +} diff --git a/serverlogic/src/main/java/com/ljsd/jieling/thread/task/PlatConfigureTask.java b/serverlogic/src/main/java/com/ljsd/jieling/thread/task/PlatConfigureTask.java index 7f6a0032d..f60864a4c 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/thread/task/PlatConfigureTask.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/thread/task/PlatConfigureTask.java @@ -1,7 +1,6 @@ package com.ljsd.jieling.thread.task; import com.ljsd.jieling.config.json.ConfMessage; -import com.ljsd.jieling.util.TimeUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; @@ -10,9 +9,7 @@ import java.io.IOException; @Component public class PlatConfigureTask implements Runnable { - private static final Logger LOGGER = LoggerFactory.getLogger(PlatConfigureTask.class); - @Override public void run() { doLogic();