断线重连

back_recharge
mashiyu 2019-02-26 16:14:22 +08:00
parent 45f52d32bd
commit 07c189f09c
6 changed files with 172 additions and 217 deletions

View File

@ -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;
}

View File

@ -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);
}
/**
* uidSession
* @param uid

View File

@ -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<Integer, BaseHandler> handlers = new HashMap<>();
public static Set<MessageTypeProto.MessageType> excludeMessges = new HashSet<>();
// public static Map<Integer, ResponseMessageCache> responseMap = new ConcurrentHashMap<>();
private static final Logger LOGGER = LoggerFactory.getLogger(ProtocolsManager.class);
private static ProtocolsManager instance = new ProtocolsManager();
private Map<Integer, BaseHandler> 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);
}
}
}

View File

@ -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<Integer, Integer> idToType = new ConcurrentHashMap();
private ConcurrentHashMap<Integer, byte[]> 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<Integer, byte[]> getIndexToMessage() {
return indexToMessage;
}
public void setIdToType(ConcurrentHashMap idToType) {
this.idToType = idToType;
public void setIndexToMessage(ConcurrentHashMap<Integer, byte[]> 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() {

View File

@ -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);

View File

@ -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 {