聊天去锁

master_0.05_aqlm
grimm 2024-03-18 13:55:50 +08:00
parent 7afdf411bf
commit a27e058066
4 changed files with 127 additions and 152 deletions

View File

@ -55,7 +55,8 @@ public class ChatLogic {
*/ */
public void getAllChatMessageInfo(ISession iSession, int chatType,long messageId) throws Exception { public void getAllChatMessageInfo(ISession iSession, int chatType,long messageId) throws Exception {
int uid = iSession.getUid(); int uid = iSession.getUid();
FriendManager friendManager = UserManager.getUser(uid).getFriendManager(); User user = UserManager.getUser(uid);
FriendManager friendManager = user.getFriendManager();
List<Integer> friends = friendManager.getFriends(); List<Integer> friends = friendManager.getFriends();
List<ChatProto.ChatInfo> chatInfoList = new ArrayList<>(); List<ChatProto.ChatInfo> chatInfoList = new ArrayList<>();
switch (chatType){ switch (chatType){
@ -66,14 +67,14 @@ public class ChatLogic {
chatInfoList = MessageCache.worldMsg.getObjs(messageId); chatInfoList = MessageCache.worldMsg.getObjs(messageId);
break; break;
case 2://公会 case 2://公会
int guildId = UserManager.getUser(uid).getPlayerInfoManager().getGuildId(); int guildId = user.getPlayerInfoManager().getGuildId();
if(guildId!=0&&MessageCache.guildMsg.containsKey(guildId)){ if(guildId!=0&&MessageCache.guildMsg.containsKey(guildId)){
chatInfoList = MessageCache.guildMsg.get(guildId).getObjs(messageId); chatInfoList = MessageCache.guildMsg.get(guildId).getObjs(messageId);
} }
break; break;
case 3: //好友 case 3: //好友
Map<Object, Object> hmget = RedisUtil.getInstence().hmget(GameApplication.serverId + RedisKey.CUser_Chat + uid); Map<Object, Object> hmget = RedisUtil.getInstence().hmget(GameApplication.serverId + RedisKey.CUser_Chat + uid);
if (hmget != null && hmget.size() != 0){ if (hmget != null && !hmget.isEmpty()){
for (Map.Entry<Object, Object> entry :hmget.entrySet()){ for (Map.Entry<Object, Object> entry :hmget.entrySet()){
String[] split = entry.getKey().toString().split("#"); String[] split = entry.getKey().toString().split("#");
String message = String.valueOf(entry.getValue()); String message = String.valueOf(entry.getValue());
@ -82,16 +83,12 @@ public class ChatLogic {
continue; continue;
} }
long time = Long.parseLong(split[1]); long time = Long.parseLong(split[1]);
User user = UserManager.getUser(userid); chatInfoList.add(CBean2Proto.getChatInfoBuilder(userid, message,time,0));
chatInfoList.add(CBean2Proto.getChatInfoBuilder(user, message,time,0));
} }
}
if (hmget!= null && !hmget.isEmpty()){
//取出数据好友聊天数据后删除 //取出数据好友聊天数据后删除
RedisUtil.getInstence().del(GameApplication.serverId + RedisKey.CUser_Chat + uid); RedisUtil.getInstence().del(GameApplication.serverId + RedisKey.CUser_Chat + uid);
} }
break; break;
case 4: //跨服 case 4: //跨服
int group = GlobleSystemLogic.getInstence().getCrossGroup(); int group = GlobleSystemLogic.getInstence().getCrossGroup();
if(group==-1){ if(group==-1){
@ -99,11 +96,10 @@ public class ChatLogic {
return; return;
} }
chatInfoList = MessageCache.getCrossMsg(messageId,String.valueOf(group)); chatInfoList = MessageCache.getCrossMsg(messageId,String.valueOf(group));
break; break;
default:{ default:
LOGGER.info("getAllChatMessageInfo wrong chatType=>{} uid=>{}", chatType, uid); LOGGER.info("getAllChatMessageInfo wrong chatType=>{} uid=>{}", chatType, uid);
} break;
} }
ChatProto.GetChatMessageInfoResponse getChatMessageInfoResponse = ChatProto.GetChatMessageInfoResponse ChatProto.GetChatMessageInfoResponse getChatMessageInfoResponse = ChatProto.GetChatMessageInfoResponse
.newBuilder().addAllChatInfo(chatInfoList) .newBuilder().addAllChatInfo(chatInfoList)
@ -153,7 +149,7 @@ public class ChatLogic {
result[i]=c; result[i]=c;
} }
tmp = new String(result); tmp = new String(result);
ChatProto.ChatInfo chatInfo = CBean2Proto.getChatInfoBuilder(user,tmp,nowTime,-1); ChatProto.ChatInfo chatInfo = CBean2Proto.getChatInfoBuilder(user.getId(),tmp,nowTime,-1);
ChatProto.SendChatInfoResponse response = ChatProto.SendChatInfoResponse.newBuilder().setChatInfo(chatInfo).build(); ChatProto.SendChatInfoResponse response = ChatProto.SendChatInfoResponse.newBuilder().setChatInfo(chatInfo).build();
MessageUtil.sendMessage(iSession,1,msgId,response,true); MessageUtil.sendMessage(iSession,1,msgId,response,true);
return; return;
@ -190,7 +186,7 @@ public class ChatLogic {
mes[i]='*'; mes[i]='*';
} }
tmp = new String(mes); tmp = new String(mes);
ChatProto.ChatInfo chatInfo = CBean2Proto.getChatInfoBuilder(user,tmp,nowTime,-1); ChatProto.ChatInfo chatInfo = CBean2Proto.getChatInfoBuilder(user.getId(),tmp,nowTime,-1);
ChatProto.SendChatInfoResponse response = ChatProto.SendChatInfoResponse.newBuilder().setChatInfo(chatInfo).build(); ChatProto.SendChatInfoResponse response = ChatProto.SendChatInfoResponse.newBuilder().setChatInfo(chatInfo).build();
MessageUtil.sendMessage(iSession,1,msgId,response,true); MessageUtil.sendMessage(iSession,1,msgId,response,true);
ReportUtil.onReportEvent(user, ReportEventEnum.SEND_MESSAGE.getType(),String.valueOf(chatType),message,new Date()); ReportUtil.onReportEvent(user, ReportEventEnum.SEND_MESSAGE.getType(),String.valueOf(chatType),message,new Date());
@ -200,7 +196,7 @@ public class ChatLogic {
switch (chatType){ switch (chatType){
case 1: //世界 case 1: //世界
long messageId = RedisUtil.getInstence().increment(GameApplication.serverId + RedisKey.CHAT_WORD_MSG_ID + GameApplication.serverId); long messageId = RedisUtil.getInstence().increment(GameApplication.serverId + RedisKey.CHAT_WORD_MSG_ID + GameApplication.serverId);
ChatProto.ChatInfo chatInfo = CBean2Proto.getChatInfoBuilder(user,message,nowTime,messageId); ChatProto.ChatInfo chatInfo = CBean2Proto.getChatInfoBuilder(user.getId(),message,nowTime,messageId);
MessageCache.addWordMsg(chatInfo,messageId); MessageCache.addWordMsg(chatInfo,messageId);
break; break;
@ -210,7 +206,7 @@ public class ChatLogic {
if(guildId==0){ if(guildId==0){
throw new ErrorCodeException(ErrorCode.CHAT_NO_FAMILY); throw new ErrorCodeException(ErrorCode.CHAT_NO_FAMILY);
} }
chatInfo = CBean2Proto.getChatInfoBuilder(user,message,nowTime,messageId); chatInfo = CBean2Proto.getChatInfoBuilder(user.getId(),message,nowTime,messageId);
MessageCache.addGuildMsg(chatInfo,messageId,guildId); MessageCache.addGuildMsg(chatInfo,messageId,guildId);
break; break;
case 3: //好友 case 3: //好友
@ -222,7 +218,7 @@ public class ChatLogic {
//现在好友推送 //现在好友推送
if (OnlineUserManager.checkUidOnline(friendId)) { if (OnlineUserManager.checkUidOnline(friendId)) {
ISession sessionByUid = OnlineUserManager.getSessionByUid(friendId); ISession sessionByUid = OnlineUserManager.getSessionByUid(friendId);
chatInfo = CBean2Proto.getChatInfoBuilder(user, message, nowTime, 0); chatInfo = CBean2Proto.getChatInfoBuilder(user.getId(), message, nowTime, 0);
sendChatInfoIndication = ChatProto.SendChatInfoIndication.newBuilder() sendChatInfoIndication = ChatProto.SendChatInfoIndication.newBuilder()
.setChatInfo(chatInfo) .setChatInfo(chatInfo)
.setType(1) .setType(1)
@ -242,7 +238,7 @@ public class ChatLogic {
} }
messageId = RedisUtil.getInstence().increment( RedisKey.CHAT_CROSS_MSG_ID + group); messageId = RedisUtil.getInstence().increment( RedisKey.CHAT_CROSS_MSG_ID + group);
chatInfo = CBean2Proto.getCrossChatInfoBuilder(user,message,nowTime,messageId); chatInfo = CBean2Proto.getChatInfoBuilder(user.getId(),message,nowTime,messageId);
MessageCache.addCrossMsg(chatInfo,messageId,String.valueOf(group)); MessageCache.addCrossMsg(chatInfo,messageId,String.valueOf(group));
break; break;

View File

@ -3,6 +3,10 @@ package com.ljsd.jieling.chat.messge;
import java.lang.reflect.Array; import java.lang.reflect.Array;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
public class LoopQueue<T> { public class LoopQueue<T> {
class Item{ class Item{
@ -19,93 +23,103 @@ public class LoopQueue<T> {
} }
} }
private int writeIndex; private final ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
private Item[] queueList; private final Lock writeLock = new ReentrantLock();
private int maxlen = 0; private int writeIndex; // 这是共享资源的索引
private int getMaxLen =0; private final Item[] queueList; // 共享资源
private long diffTime ; private int maxlen; // 假设的最大长度
private int initLength =20; private long diffTime;// 时间差阈值,假设
private int initLength = 20; // 假设的初始长度
public LoopQueue(int maxlen, int getMaxLen, long diffTime) { public LoopQueue(int maxlen, long diffTime) {
this.maxlen = maxlen; this.maxlen = maxlen;
queueList = (Item[])Array.newInstance(Item.class, maxlen); queueList = (Item[])Array.newInstance(Item.class, maxlen);
for(int i=0; i<maxlen; i++){ for(int i=0; i<maxlen; i++){
queueList[i] = new Item(); queueList[i] = new Item();
} }
writeIndex = 0; writeIndex = 0;
this.getMaxLen = getMaxLen;
this.diffTime = diffTime; this.diffTime = diffTime;
} }
//添加一个新的元素到循环队列中 //添加一个新的元素到循环队列中
public synchronized void push(T obj,long messageId) { public void push(T obj, long messageId) {
this.queueList[writeIndex%maxlen].setObj(obj,messageId); // writeLock.lock();
writeIndex = writeIndex+1; try {
int index = writeIndex % maxlen;
// 假设 setObj 是一个正确设置对象和消息 ID 的方法
queueList[index].setObj(obj,messageId); // 或者使用 setObj 方法,如果它存在
writeIndex++;
} finally {
// writeLock.unlock();
}
} }
//diffTime 允许的取到的最小时间差的数据,为-1时则不校验时间差 //diffTime 允许的取到的最小时间差的数据,为-1时则不校验时间差
//getMaxLen 允许取到的最大数量 //getMaxLen 允许取到的最大数量
public synchronized List<T> getObjs(long messageId) { public List<T> getObjs(long messageId) {
List<T> objs = new ArrayList<>(); List<T> objs = new ArrayList<>();
int nextIndex = writeIndex%maxlen-1;
int endIndex = writeIndex%maxlen;
if(nextIndex==-1){
nextIndex = maxlen - 1;
}
boolean needCompareTime = (messageId == 0); boolean needCompareTime = (messageId == 0);
long curTime = System.currentTimeMillis(); // readWriteLock.readLock().lock();
while(queueList[nextIndex].obj != null && nextIndex != endIndex){ try {
int nextIndex = (writeIndex - 1 + maxlen) % maxlen;
int endIndex = writeIndex % maxlen;
if(!needCompareTime){ long curTime = System.currentTimeMillis();
//比较消息id
if(messageId>=queueList[nextIndex].messgeId){ while (queueList[nextIndex] != null && nextIndex != endIndex) {
if (!needCompareTime && messageId >= queueList[nextIndex].messgeId) {
break; break;
} }
} if (needCompareTime) {
if (objs.size() >= initLength ||
(diffTime != -1 && (curTime - queueList[nextIndex].timestamp) >= diffTime)) {
break;
}
}
if (needCompareTime && objs.size() >= initLength ){ objs.add(0, queueList[nextIndex].obj);
break;
}
if(needCompareTime){ if (objs.size() > maxlen) {
if(diffTime!=-1&& (curTime - queueList[nextIndex].timestamp) >= diffTime){
break; break;
} }
}
objs.add(0, queueList[nextIndex].obj);
if(objs.size()>maxlen){
break;
}
nextIndex--; nextIndex = (nextIndex - 1 + maxlen) % maxlen;
if(nextIndex == -1){
nextIndex = maxlen - 1;
} }
} finally {
// readWriteLock.readLock().unlock();
} }
if(!needCompareTime && objs.size()>getMaxLen){
return objs.subList(0,getMaxLen); // 后处理不需要在锁中进行
if (!needCompareTime && objs.size() > maxlen) {
return objs.subList(0, maxlen);
} }
return objs; return objs;
} }
public synchronized String toString() { @Override
StringBuilder builder = new StringBuilder(); public String toString() {
for(int i=0; i<maxlen; i++){ // readWriteLock.readLock().lock();
builder.append(i); try {
if(i == writeIndex%maxlen){ StringBuilder builder = new StringBuilder();
builder.append("(writeIndex)"); for (int i = 0; i < maxlen; i++) {
builder.append(i);
if (i == writeIndex % maxlen) {
builder.append("(writeIndex)");
}
builder.append("->");
if (queueList[i] == null) {
builder.append("-1");
} else {
builder.append(queueList[i].toString()); // 确保obj有有效的toString方法
}
builder.append(",");
} }
builder.append("->"); return builder.toString();
if(queueList[i].obj == null){ } finally {
builder.append("-1"); // readWriteLock.readLock().unlock();
} else {
builder.append(queueList[i].obj);
}
builder.append(",");
} }
return builder.toString();
} }
} }

View File

@ -3,23 +3,18 @@ package com.ljsd.jieling.chat.messge;
import com.google.gson.Gson; import com.google.gson.Gson;
import com.ljsd.jieling.db.redis.RedisKey; import com.ljsd.jieling.db.redis.RedisKey;
import com.ljsd.jieling.db.redis.RedisUtil; import com.ljsd.jieling.db.redis.RedisUtil;
import org.luaj.vm2.ast.Str;
import rpc.protocols.ChatProto; import rpc.protocols.ChatProto;
import java.util.ArrayList; import java.util.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
public class MessageCache { public class MessageCache {
private static final int capuatiy =30; private static final int capuatiy =30;
private static final int msgInterval = 24*3600*1000; private static final int msgInterval = 24*3600*1000;
private static final int perCaught= 30; private static final int perCaught= 30;
public static LoopQueue worldMsg = new LoopQueue(capuatiy,perCaught,-1); public static LoopQueue worldMsg = new LoopQueue(capuatiy, -1);
public static LoopQueue sysMsg = new LoopQueue(capuatiy,perCaught,msgInterval); public static LoopQueue sysMsg = new LoopQueue(capuatiy, msgInterval);
public static Map<Integer,LoopQueue> guildMsg = new ConcurrentHashMap(); public static Map<Integer,LoopQueue> guildMsg = new ConcurrentHashMap();
private static Gson gson = new Gson(); private static Gson gson = new Gson();
@ -32,7 +27,7 @@ public class MessageCache {
} }
public static void addGuildMsg(ChatProto.ChatInfo chatInfo,long messageId,int guildId){ public static void addGuildMsg(ChatProto.ChatInfo chatInfo,long messageId,int guildId){
if(!guildMsg.containsKey(guildId)){ if(!guildMsg.containsKey(guildId)){
guildMsg.put(guildId,new LoopQueue(capuatiy,perCaught,-1)); guildMsg.put(guildId,new LoopQueue(capuatiy, -1));
} }
guildMsg.get(guildId).push(chatInfo,messageId); guildMsg.get(guildId).push(chatInfo,messageId);
} }
@ -44,53 +39,41 @@ public class MessageCache {
RedisUtil.getInstence().hset(key,String.valueOf(index),gson.toJson(chatInfo)); RedisUtil.getInstence().hset(key,String.valueOf(index),gson.toJson(chatInfo));
} }
public static List<ChatProto.ChatInfo> getCrossMsg(long messageId,String group){ public static List<ChatProto.ChatInfo> getCrossMsg(long messageId, String group) {
String key = RedisKey.CHAT_CROSS_INFO_CACHE+group; String key = RedisKey.CHAT_CROSS_INFO_CACHE + group;
Map<Integer,ChatProto.ChatInfo> map = new HashMap<>(); Map<Object, Object> rawMap = RedisUtil.getInstence().hmget(key);
Map<Object, Object> collect = RedisUtil.getInstence().hmget(key);
collect.entrySet().forEach(o -> map.put(Integer.parseInt(o.getKey().toString()), if (rawMap == null || rawMap.isEmpty()) {
gson.fromJson(o.getValue().toString(),ChatProto.ChatInfo.class))); return Collections.emptyList();
}
List<ChatProto.ChatInfo> objs = new ArrayList<>(); List<ChatProto.ChatInfo> objs = new ArrayList<>();
Object obj = RedisUtil.getInstence().get(RedisKey.CHAT_CROSS_MSG_ID + group); Object obj = RedisUtil.getInstence().get(RedisKey.CHAT_CROSS_MSG_ID + group);
if(obj==null){
if (obj == null) {
return objs; return objs;
} }
int writeIndex = Integer.parseInt(String.valueOf(obj)); int writeIndex = Integer.parseInt(String.valueOf(obj));
int nextIndex = writeIndex % capuatiy; // Assuming capacity is a static value or defined elsewhere
Integer nextIndex = writeIndex%capuatiy;
if(nextIndex==-1){
nextIndex = capuatiy - 1;
}
if(collect.size()<=0){
return objs;
}
boolean needCompareTime = (messageId == 0); boolean needCompareTime = (messageId == 0);
while(map.containsKey(nextIndex)){ while (rawMap.containsKey(nextIndex)) {
String rawValue = String.valueOf(rawMap.get(nextIndex));
ChatProto.ChatInfo chatInfo = gson.fromJson(rawValue, ChatProto.ChatInfo.class);
ChatProto.ChatInfo o = map.get(nextIndex); if (!needCompareTime && messageId >= chatInfo.getMessageId()) {
if(!needCompareTime){
//比较消息id
if(messageId>=o.getMessageId()){
break;
}
}
objs.add(0, o);
if(objs.size()>perCaught){
break; break;
} }
nextIndex--;
if(nextIndex == -1){
nextIndex = capuatiy - 1;
}
}
if(!needCompareTime && objs.size()>perCaught){
return objs.subList(0,perCaught);
}
return objs; objs.add(0, chatInfo);
if (objs.size() > perCaught) {
break;
}
nextIndex = (nextIndex - 1 + capuatiy) % capuatiy;
}
return objs.size() > perCaught ? objs.subList(0, perCaught) : objs;
} }
} }

View File

@ -786,42 +786,24 @@ public class CBean2Proto {
.build(); .build();
} }
public static ChatProto.ChatInfo getChatInfoBuilder(int uid,String message,long time,long messageId) {
public static ChatProto.ChatInfo getChatInfoBuilder(User user,String message,long time,long messageId) {
ChatProto.ChatInfo.Builder chatInfo = ChatProto.ChatInfo.newBuilder(); ChatProto.ChatInfo.Builder chatInfo = ChatProto.ChatInfo.newBuilder();
PlayerManager playerInfoManager = user.getPlayerInfoManager(); CSPlayer csPlayer = CrossServiceLogic.getPlayerByRedis(uid);
chatInfo.setSenderId(user.getId()); if (csPlayer != null){
chatInfo.setSenderName(playerInfoManager.getNickName()); chatInfo.setSenderId(csPlayer.getUserId());
chatInfo.setSenderlevel(playerInfoManager.getLevel()); chatInfo.setSenderName(csPlayer.getName());
chatInfo.setSendervip(playerInfoManager.getVipLevel()); chatInfo.setSenderlevel(csPlayer.getLevel());
chatInfo.setTimes(time); chatInfo.setTimes(time);
chatInfo.setMsg(message); chatInfo.setMsg(message);
chatInfo.setHead(playerInfoManager.getHead()); chatInfo.setHead(csPlayer.getHead());
chatInfo.setFrame(playerInfoManager.getHeadFrame()); chatInfo.setFrame(csPlayer.getHeadFrame());
chatInfo.setSoulVal(playerInfoManager.getMaxForce()); chatInfo.setSoulVal(csPlayer.getMaxFore());
chatInfo.setMessageId(messageId); chatInfo.setMessageId(messageId);
chatInfo.setUserTitle(playerInfoManager.getUserTitle()); chatInfo.setUserTitle(csPlayer.getUserTitle());
chatInfo.setPracticeLevel(user.getHeroManager().getPracticeLevel()); chatInfo.setPracticeLevel(csPlayer.getPracticeLevel());
return chatInfo.build(); String serverName = CrossServiceLogic.getServerName(csPlayer.getServerId());
} chatInfo.setServerName(serverName);
}
public static ChatProto.ChatInfo getCrossChatInfoBuilder(User user,String message,long time,long messageId){
ChatProto.ChatInfo.Builder chatInfo = ChatProto.ChatInfo.newBuilder();
PlayerManager playerInfoManager = user.getPlayerInfoManager();
chatInfo.setSenderId(user.getId());
chatInfo.setSenderName(playerInfoManager.getNickName());
chatInfo.setSenderlevel(playerInfoManager.getLevel());
chatInfo.setSendervip(playerInfoManager.getVipLevel());
chatInfo.setTimes(time);
chatInfo.setMsg(message);
chatInfo.setHead(playerInfoManager.getHead());
chatInfo.setFrame(playerInfoManager.getHeadFrame());
chatInfo.setSoulVal(playerInfoManager.getMaxForce());
chatInfo.setMessageId(messageId);
chatInfo.setUserTitle(playerInfoManager.getUserTitle());
chatInfo.setPracticeLevel(user.getHeroManager().getPracticeLevel());
String serverName = CrossServiceLogic.simplifyServerName(GameApplication.serverId);
chatInfo.setServerName(serverName);
return chatInfo.build(); return chatInfo.build();
} }