统一线程处理数据
parent
2d11d1485e
commit
745ddbeadf
|
@ -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<String, Map<String, MongoUpdateCache.UpdateRequest>> 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();
|
||||
|
|
|
@ -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<String, Map<String, UpdateRequest>> requestMap = new HashMap<>();
|
||||
static final BlockingQueue<Map<String, Map<String, UpdateRequest>>> logQueue = new BlockingUniqueQueue<>();
|
||||
private final Map<String, Map<String, UpdateRequest>> updateDataMap = new HashMap<>();
|
||||
|
||||
protected void addUpdateRequest(MongoBase mongoBase, String fullKey, Object value, int opCode) {
|
||||
String key = mongoBase.getRootCollection() + ":" + mongoBase.getRootId();
|
||||
Map<String, UpdateRequest> listRequest = requestMap.get(key);
|
||||
if (listRequest == null) {
|
||||
listRequest = new HashMap<>();
|
||||
requestMap.put(key, listRequest);
|
||||
} else {
|
||||
Iterator<Map.Entry<String, UpdateRequest>> it = listRequest.entrySet().iterator();
|
||||
while (it.hasNext()) {
|
||||
Map.Entry<String, UpdateRequest> 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<String, Map<String, UpdateRequest>> dataMap = requestMap;
|
||||
if (MongoKeys.mongoKeyMap.containsKey(fullKey.split("\\.")[0]) || MongoKeys.mongoKeyMap.containsKey(fullKey)) {
|
||||
dataMap = updateDataMap;
|
||||
}
|
||||
Map<String, UpdateRequest> listRequest = dataMap.get(key);
|
||||
if (listRequest == null) {
|
||||
listRequest = new HashMap<>();
|
||||
dataMap.put(key, listRequest);
|
||||
} else {
|
||||
Iterator<Map.Entry<String, UpdateRequest>> it = listRequest.entrySet().iterator();
|
||||
while (it.hasNext()) {
|
||||
Map.Entry<String, UpdateRequest> 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<String, Map<String, UpdateRequest>> updateRequestMap) {
|
||||
updateMongoData(ljsdMongoTemplate, updateRequestMap);
|
||||
}
|
||||
|
||||
private void updateMongoData(LjsdMongoTemplate ljsdMongoTemplate, Map<String, Map<String, UpdateRequest>> updateRequestMap) {
|
||||
BasicDBObject searchQuery = new BasicDBObject();
|
||||
|
||||
for (Map.Entry<String, Map<String, UpdateRequest>> entry : requestMap.entrySet()) {
|
||||
for (Map.Entry<String, Map<String, UpdateRequest>> 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();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
@ -1,12 +1,11 @@
|
|||
package com.ljsd.common.mogodb;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
public class MongoUpdateCacheThreadLocal {
|
||||
private final static ThreadLocal<MongoUpdateCache> 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<String, Map<String, MongoUpdateCache.UpdateRequest>> updateRequestMap) {
|
||||
MongoUpdateCache mongoUpdateCache = getMongoUpdateCache();
|
||||
mongoUpdateCache.concurrentDataUpdate(ljsdMongoTemplate, updateRequestMap);
|
||||
}
|
||||
}
|
||||
|
|
|
@ -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<Object> dataQueue = new BlockingUniqueQueue<>();
|
||||
static BlockingQueue<Map<String, Map<String, MongoUpdateCache.UpdateRequest>>> dataQueue = new BlockingUniqueQueue<>();
|
||||
//入队列
|
||||
public static void addLogQueue(Map<String, Map<String, MongoUpdateCache.UpdateRequest>> info){
|
||||
dataQueue.offer(info);
|
||||
}
|
||||
|
||||
public static void manageData(LjsdMongoTemplate myMongoTemplate) throws Exception {
|
||||
Map<String, Map<String, MongoUpdateCache.UpdateRequest>> take = ConcurrentDataMessage.dataQueue.take();
|
||||
myMongoTemplate.concurrentDataUpdate(take);
|
||||
}
|
||||
}
|
|
@ -0,0 +1,13 @@
|
|||
package com.ljsd.common.mogodb.util;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
public class MongoKeys {
|
||||
|
||||
public static Map<String, Integer> mongoKeyMap = new HashMap<>();
|
||||
|
||||
public static void initmongoKey() {
|
||||
mongoKeyMap.put("friendManager",1);
|
||||
}
|
||||
}
|
|
@ -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");
|
||||
|
|
|
@ -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 {
|
||||
|
||||
|
|
|
@ -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 () {
|
||||
|
|
|
@ -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() {
|
||||
|
|
|
@ -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);
|
||||
}
|
||||
}
|
||||
}
|
|
@ -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();
|
||||
|
|
Loading…
Reference in New Issue