From 07c189f09c13edd2e3315108301fd0e5b1dfd255 Mon Sep 17 00:00:00 2001 From: mashiyu Date: Tue, 26 Feb 2019 16:14:22 +0800 Subject: [PATCH] =?UTF-8?q?=E6=96=AD=E7=BA=BF=E9=87=8D=E8=BF=9E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../com/ljsd/jieling/db/redis/RedisUtil.java | 4 +- .../ljsd/jieling/logic/OnlineUserManager.java | 13 + .../network/server/ProtocolsManager.java | 325 +++++++----------- .../jieling/network/session/ISession.java | 29 +- .../java/com/ljsd/jieling/util/MathUtils.java | 3 + .../com/ljsd/jieling/util/MessageUtil.java | 15 +- 6 files changed, 172 insertions(+), 217 deletions(-) diff --git a/serverlogic/src/main/java/com/ljsd/jieling/db/redis/RedisUtil.java b/serverlogic/src/main/java/com/ljsd/jieling/db/redis/RedisUtil.java index 389dc4602..14beb218c 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/db/redis/RedisUtil.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/db/redis/RedisUtil.java @@ -113,8 +113,8 @@ public class RedisUtil { return key == null ? null : redisTemplate.opsForValue().get(key); } - public String getAndDelete(String key) { - String value = redisTemplate.opsForValue().get(key); + public Object getAndDelete(String key) { + Object value = redisTemplate.opsForValue().get(key); del(key); return value; } diff --git a/serverlogic/src/main/java/com/ljsd/jieling/logic/OnlineUserManager.java b/serverlogic/src/main/java/com/ljsd/jieling/logic/OnlineUserManager.java index 23083ccee..2cab3934f 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/logic/OnlineUserManager.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/logic/OnlineUserManager.java @@ -21,6 +21,19 @@ public class OnlineUserManager { sessionMap.remove(uid); } + + public static void addPlayerToMap(int uid, PlayerManager player) { + onlineUserMap.putIfAbsent(uid,player); + } + + public static void removePlayerFromMap(int uid) { + onlineUserMap.remove(uid); + } + + public static PlayerManager getPlayerFromMap(int uid) { + return onlineUserMap.get(uid); + } + /** * 通过uid获取Session * @param uid diff --git a/serverlogic/src/main/java/com/ljsd/jieling/network/server/ProtocolsManager.java b/serverlogic/src/main/java/com/ljsd/jieling/network/server/ProtocolsManager.java index 6a5d7ebd3..de75c507c 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/network/server/ProtocolsManager.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/network/server/ProtocolsManager.java @@ -1,8 +1,10 @@ package com.ljsd.jieling.network.server; +import com.google.protobuf.GeneratedMessage; import com.google.protobuf.InvalidProtocolBufferException; -import com.google.protobuf.MessageLite; import com.ljsd.GameApplication; +import com.ljsd.jieling.db.redis.RedisKey; +import com.ljsd.jieling.db.redis.RedisUtil; import com.ljsd.jieling.exception.ErrorCode; import com.ljsd.jieling.handler.BaseHandler; import com.ljsd.jieling.logic.OnlineUserManager; @@ -12,31 +14,29 @@ import com.ljsd.jieling.network.session.ISession; import com.ljsd.jieling.protocol.ProtocolsAbstract; import com.ljsd.jieling.protocols.MessageTypeProto; import com.ljsd.jieling.session.GameSession; +import com.ljsd.jieling.util.MathUtils; import com.ljsd.jieling.util.MessageUtil; -import io.netty.buffer.ByteBuf; -import io.netty.buffer.ByteBufHolder; -import io.netty.channel.group.ChannelGroup; -import io.netty.util.ReferenceCountUtil; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.stereotype.Component; -import java.util.*; +import java.util.HashMap; +import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; /** - * Created by justin on 14-7-30. + * * 消息类型-消息处理 管理 * 在新加一个消息时,都要在些处添加对应关系 */ @Component public class ProtocolsManager implements ProtocolsAbstract { - private static final Logger LOGGER = LoggerFactory.getLogger(ProtocolsManager.class); - private static ProtocolsManager instance = new ProtocolsManager(); - private Map handlers = new HashMap<>(); - public static Set excludeMessges = new HashSet<>(); -// public static Map responseMap = new ConcurrentHashMap<>(); + private static final Logger LOGGER = LoggerFactory.getLogger(ProtocolsManager.class); + + private static ProtocolsManager instance = new ProtocolsManager(); + + private Map handlers = new HashMap<>(); private ProtocolsManager() { } @@ -45,118 +45,10 @@ public class ProtocolsManager implements ProtocolsAbstract { return instance; } - /** - * 注册处理方法,在应用启动时调用一次 - */ -// public final int HANDLER_THREAD_NUM = 100; -// -// public HandlerLogicThread[] handlerThreads = new HandlerLogicThread[HANDLER_THREAD_NUM]; -// -// -// public void initContext() { -// for (int i = 0; i < HANDLER_THREAD_NUM; i++) { -// handlerThreads[i] = new HandlerLogicThread(); -// handlerThreads[i].setPriority(9);//优先级最高 -// handlerThreads[i].setName("h-" + i); -// handlerThreads[i].start(); -// } -// } -// -// public ResponseMessageCache getResponseCache(int uid) { -// ResponseMessageCache cache = responseMap.get(uid); -// if (cache == null) { -// cache = new ResponseMessageCache(); -// responseMap.put(uid, cache); -// } -// return cache; -// } - /** - * 所有发消息的地方调用此方法 - * - * @param session - * @param gameMessage - * @param flush - */ - public void sendMessage(ISession session, MessageLite gameMessage, boolean flush, boolean isResponse) { - - if (flush && isResponse ) { - session.writeAndFlush(gameMessage); - } else { - session.write(gameMessage); - } - } - -// public void cleanResponseCache(int uid, boolean save) { -// try { -// LOGGER.info("cleanResponseCache->uid={},isClean={}", uid, save); -// ResponseMessageCache cache = responseMap.get(uid); -// if (cache == null) { -// return; -// } -// ResponseMessageCache.clean(cache, uid, save); -// responseMap.remove(uid); -// } catch (Exception e) { -// LOGGER.error("cleanResponseCache->uid={},msg={}", uid, e.getMessage(), e); -// } -// } -// -// public void reloadResponseCache(int uid) { -// try { -// ResponseMessageCache cache = responseMap.get(uid); -// if (cache == null) { -// cache = ResponseMessageCache.reload(uid); -// if (cache != null) { -// responseMap.put(uid, cache); -// } -// } -// if (cache == null) { -// getResponseCache(uid); -// } -// } catch (Exception e) { -// LOGGER.error("reloadResponseCache->uid={},msg={}", uid, e.getMessage(), e); -// } -// } - - private static Object safeDuplicate(Object message) { - if (message instanceof ByteBuf) { - return ((ByteBuf) message).duplicate().retain(); - } else if (message instanceof ByteBufHolder) { - return ((ByteBufHolder) message).duplicate().retain(); - } else { - return ReferenceCountUtil.retain(message); - } - } private AtomicInteger msgCount = new AtomicInteger(0); - public void sendAllUser(ChannelGroup group, MessageLite gameMessage, boolean flush) { - group.write(gameMessage); - int andIncrement = msgCount.getAndIncrement(); - if (andIncrement % 20 == 0) { - group.flush(); - } - } - - /** - * 给指定用户发消息 - * - * @param uid - * @param gameMessage - * @param flush - */ - public void sendMessageByUid(int uid, MessageLite gameMessage, boolean flush) { - ISession session = (ISession) GameMessageHandler.currentSessions.get(uid); - if (session != null) { - if (flush) { - session.writeAndFlush(gameMessage); - } else { - session.write(gameMessage); - } - } else { - LOGGER.debug("sendMessageByUid->uid={};session is null", uid); - } - } public void addHandler(BaseHandler handler) { handlers.put(handler.getMessageCode().getNumber(), handler); @@ -199,17 +91,40 @@ public class ProtocolsManager implements ProtocolsAbstract { final ISession session = (ISession) gameSession; PacketNetData packetNetData = new PacketNetData((byte[])obj); - session.setUid(packetNetData.getUserId()); - session.setToken(packetNetData.getToken()); - if (packetNetData.getMsgId()==1000){ dealHeartBeat(session); return; } - int resultCode = checkAndRefresh(session,packetNetData); - if (resultCode != 0 ){ - MessageUtil.sendErrorCode(session,0); + if (packetNetData.getMsgId()== MessageTypeProto.MessageType.LOGIN_REQUEST_VALUE || + packetNetData.getMsgId()== MessageTypeProto.MessageType.RECONNECT_REQUEST_VALUE){ + + //初始化session + if (sessionInit(session,packetNetData)){ + MessageUtil.sendErrorCode(session,0,"您的账号token不正确,请退出游戏重新登陆!"); + return; + } + //处理多设备重登 + if (dealRepeatLogin(session)==1){ + MessageUtil.sendErrorCode(session,0,"您的账号已经在其他设备上登陆,请确认账号安全!"); + return; + } + //刷新token + refreshToken(session); + //在线列表加入session + OnlineUserManager.userOnline(session.getUid(),session); + } + //处理心跳 + if (packetNetData.getMsgId()==1000){ + dealHeartBeat(session); + return; + } + //检查token + if (checkToken(session,packetNetData) != 0 ){ + MessageUtil.sendErrorCode(session,0,"您的账号token不正确,请退出游戏重新登陆!"); + return; + } + if (checkIndex(session,packetNetData) != 0){ return; } @@ -228,53 +143,63 @@ public class ProtocolsManager implements ProtocolsAbstract { } } + private void refreshToken(ISession session) { + int token = Math.abs(MathUtils.randomInt()); + session.setToken(token); + session.setUserLoginTime(System.currentTimeMillis()/1000); + + } + + private boolean sessionInit(ISession session, PacketNetData packetNetData) { + int uid = packetNetData.getUserId(); + int token = packetNetData.getToken(); + session.setUid(uid); + String tokenKey = RedisKey.getKey(RedisKey.TOKEN, String.valueOf(uid),false); + int serverToken = (int)RedisUtil.getInstence().get(tokenKey); + if (token == serverToken){ + session.setToken(token); + return true; + } + return false; + } + + + private int dealRepeatLogin(ISession session) { + if (OnlineUserManager.checkUidOnline(session.getUid())){ + ISession iSession = OnlineUserManager.getSessionByUid(session.getUid()); + MessageUtil.sendErrorCode(iSession, ErrorCode.reloginCode,"您的账号已经在其他设备上登陆,请确认账号安全!"); + OnlineUserManager.userOffline(iSession.getUid()); + iSession.close(); + return 1; + } + return 0; + } + + private int checkIndex(ISession iSession, PacketNetData packetNetData) { + int index = packetNetData.getIndex(); + if (iSession.getIndex() >= index){ + byte[] message = iSession.getCacheMessage(index); + if (message==null && packetNetData.getMsgId() != 11100){ + //TODO error + }else { + iSession.writeAndFlush(message); + return 1; + } + }else { + iSession.setIndex(index); + iSession.setLoginTime(System.currentTimeMillis()); + } + return 0; + } + private void dealHeartBeat(ISession session) { -// new Thread(new Runnable() { -// @Override -// public void run() { -// -// Random random = new Random(); -// int time = random.nextInt(6)+1; -// try { -// System.out.println("sleep time : "+time); -// Thread.sleep(time*1000); -// } catch (InterruptedException e) { -// e.printStackTrace(); -// } -// -// Thread.currentThread().setName(""); -// MessageUtil.sendHeartBeat(session); -// } -// }).start(); - //TODO MessageUtil.sendHeartBeat(session); } - private int checkAndRefresh(ISession session, PacketNetData packetNetData) { - int myToken = 0; - if (packetNetData.getMsgId()== MessageTypeProto.MessageType.LOGIN_REQUEST_VALUE){ - - if (OnlineUserManager.checkUidOnline(session.getUid())){ - ISession iSession = OnlineUserManager.getSessionByUid(session.getUid()); - MessageUtil.sendErrorCode(session, ErrorCode.reloginCode); - OnlineUserManager.userOffline(iSession.getUid()); - iSession.close(); - } - OnlineUserManager.userOnline(session.getUid(),session); - - //TODO 更新token - }else if (packetNetData.getMsgId()== MessageTypeProto.MessageType.RECONNECT_REQUEST_VALUE){ - //TODO 更新token - //TODO - }else { - + private int checkToken(ISession session, PacketNetData packetNetData) { + if (session.getToken() != packetNetData.getToken()){ + return 1; } - -// if (myToken != packetNetData.getToken()){ -// return 1; -// } - session.setUserLoginTime(System.currentTimeMillis()/1000); - session.setToken(myToken); return 0; } @@ -294,17 +219,6 @@ public class ProtocolsManager implements ProtocolsAbstract { return; } -// ISession nowSession = (ISession) GameMessageHandler.currentSessions.get(session.getUid()); -// if (nowSession == null) { -// LOGGER.info("offLine->uid={},session={},, nowSession is null", -// session.getUid(), session); -// return; -// } -// if (nowSession != null && !session.getId().equals(nowSession.getId())) { -// LOGGER.error("offLine->uid={},nowUid={}", session.getUid(), nowSession.getId()); -// return; -// } - GameMessageHandler.currentSessions.remove(session.getUid()); try { @@ -317,33 +231,56 @@ public class ProtocolsManager implements ProtocolsAbstract { } public boolean kickOldUser(int uid, String uToken, int errorCode, String errorMsg, long requestTime) { + if (GameMessageHandler.currentSessions.size() != 0 && GameMessageHandler.currentSessions.keySet().contains(uid)) { + if (requestTime != 0 && requestTime < GameMessageHandler.currentSessions.get(uid).getUserLoginTime()) { + LOGGER.info("kickOldUser->uid={},requestTime={},userLoginTime={}", + uid, requestTime, GameMessageHandler.currentSessions.get(uid).getUserLoginTime()); + return false; + } + } + if (!OnlineUserManager.checkUidOnline(uid)) { + LOGGER.info("kickOldUser->uid={};isNotOnline", uid); + return false; + } + ISession oldSession = OnlineUserManager.getSessionByUid(uid); + if (oldSession == null) { + LOGGER.info("kickOldUser->uid={};oldSessionisNull", uid); + OnlineUserManager.userOffline(uid); + return false; + } + oldSession.setOfflineType(errorCode); + +// if (oldSession.getCtx() != null && !uToken.equals(oldSession.getToken())) { +// Channel oldChannl = oldSession.getCtx().channel(); +// if (errorCode == ErrorCode.LOGIN_IN_OTHER_GAMESERVER) { +// OnlineUserLogics.removeFromOnline(uid); +// //转化为踢人消息, +// // 如果当前设备重新登录或重连,登录的其他服务器 +// // 那么当前连接应该已经断开,客户端不会收到该消息,则当前设备不会被踢下线 +// //如果用其他设备登录,则当前设备能够收到这条消息,则当前设备被踢下线 +// errorCode = ErrorCode.LOGIN_IN_OTHER; +// } +// LOGGER.info("kickOldUser->uid={},errorCode={};sendErrorMessageByChannel!", uid, errorCode); +// sendErrorMessageByChannel(oldChannl, errorCode, errorMsg); +// } + + offLine(oldSession); +// oldSession.close(); return true; } @Override public void readIdel(GameSession gameSession) { // TODO -// final ISession session = (ISession) gameSession; -// readIdea(session); + final ISession session = (ISession) gameSession; + readIdea(session); } public BaseHandler getHandler(MessageTypeProto.MessageType messageType) { return handlers.get(messageType); } - public void sendChatMessage(ISession session, MessageLite gameMessage) { - try { - if (session != null && session.getCtx() != null) { - if (session.getCtx().channel().isWritable()) { - session.getCtx().writeAndFlush(gameMessage); - } else { - session.getCtx().writeAndFlush(gameMessage).sync(); - } - } - } catch (Exception e) { - LOGGER.error("sendChatMessage->uid={},msg={}", session.getUid(), e.getMessage(), e); - } - } + } diff --git a/serverlogic/src/main/java/com/ljsd/jieling/network/session/ISession.java b/serverlogic/src/main/java/com/ljsd/jieling/network/session/ISession.java index b004eb688..60fbdcf1e 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/network/session/ISession.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/network/session/ISession.java @@ -1,5 +1,7 @@ package com.ljsd.jieling.network.session; +import com.google.protobuf.GeneratedMessage; import com.ljsd.jieling.netty.NettyGameSession; +import com.ljsd.jieling.network.server.ServerCacheMessage; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -12,7 +14,7 @@ public class ISession extends NettyGameSession { private boolean kick; private long loginTime;//BI用 - private ConcurrentHashMap idToType = new ConcurrentHashMap(); + private ConcurrentHashMap indexToMessage = new ConcurrentHashMap(); private int heartBreatNums = 3; @@ -61,28 +63,21 @@ public class ISession extends NettyGameSession { this.kick = kick; } - public ConcurrentHashMap getIdToType() { - return idToType; + public ConcurrentHashMap getIndexToMessage() { + return indexToMessage; } - public void setIdToType(ConcurrentHashMap idToType) { - this.idToType = idToType; + public void setIndexToMessage(ConcurrentHashMap indexToMessage) { + this.indexToMessage = indexToMessage; } - public void addIdToTypeMap(int messageIndex, int typeId) { - this.idToType.put(typeId, messageIndex); + public void putBackMassageToMap(byte[] backMessage){ + this.indexToMessage.clear(); + this.indexToMessage.putIfAbsent(this.index,backMessage); } - public int getRequestIndex(int requestTypeId) { - Integer requestIndex = this.idToType.get(requestTypeId); - if (requestIndex != null) { - return requestIndex; - } - return 0; - } - - public void removeRequestTypeId(int requestTypeId) { - this.idToType.remove(requestTypeId); + public byte[] getCacheMessage(int index){ + return this.indexToMessage.get(index); } public long getLoginTime() { diff --git a/serverlogic/src/main/java/com/ljsd/jieling/util/MathUtils.java b/serverlogic/src/main/java/com/ljsd/jieling/util/MathUtils.java index 874a47262..87ff8a10f 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/util/MathUtils.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/util/MathUtils.java @@ -89,6 +89,9 @@ public class MathUtils { return random.nextInt(n); } + public static int randomInt(){ + return random.nextInt(); + } public static int numRount(float num){ return Math.round(num); diff --git a/serverlogic/src/main/java/com/ljsd/jieling/util/MessageUtil.java b/serverlogic/src/main/java/com/ljsd/jieling/util/MessageUtil.java index 20afaf28f..6f68d2aec 100644 --- a/serverlogic/src/main/java/com/ljsd/jieling/util/MessageUtil.java +++ b/serverlogic/src/main/java/com/ljsd/jieling/util/MessageUtil.java @@ -5,6 +5,7 @@ import com.ljsd.jieling.db.redis.RedisKey; import com.ljsd.jieling.logic.dao.UserManager; import com.ljsd.jieling.netty.PackageConstant; import com.ljsd.jieling.netty.cocdex.Tea; +import com.ljsd.jieling.network.server.ServerCacheMessage; import com.ljsd.jieling.network.session.ISession; import com.ljsd.jieling.protocols.CommonProto; import com.ljsd.jieling.protocols.MessageTypeProto; @@ -77,13 +78,19 @@ public class MessageUtil { session.writeAndFlush(byteBuf); } - public static void sendErrorCode(ISession session,int errorCode) { + //服务器推送错误码,一般情况是系统错误 + public static void sendErrorCode(ISession session,int errorCode, String errMsg) { + if( null == errMsg){ + errMsg=""; + } + GeneratedMessage generatedMessage = CommonProto.ErrorResponse.newBuilder().setErrCode(errorCode).setErrMsg(errMsg).build(); byte[] byteBuf = wrappedBuffer(session.getUid(), session.getToken(), session.getIndex(), - errorCode, MessageTypeProto.MessageType.ERROR_CODE_INDICATION_VALUE, null); + errorCode, MessageTypeProto.MessageType.ERROR_CODE_INDICATION_VALUE, generatedMessage); session.writeAndFlush(byteBuf); } - public static void sendErrorResponse(ISession session,int errorCode,int msgId,String errMsg) throws Exception { + //服务器逻辑错误返回,一般情况是业务逻辑错误 + public static void sendErrorResponse(ISession session,int errorCode,int msgId, String errMsg) throws Exception { GeneratedMessage generatedMessage = null; if( null == errMsg){ errMsg=""; @@ -93,7 +100,7 @@ public class MessageUtil { byte[] byteBuf = wrappedBuffer(session.getUid(), session.getToken(), session.getIndex(), errorCode,msgId, generatedMessage); session.writeAndFlush(byteBuf); - + session.putBackMassageToMap(byteBuf); } // // public static void sendResponse(Msg msg, int msgId, GeneratedMessage generatedMessage) throws Exception {