back_recharge
jiahuiwen 2019-02-25 20:43:01 +08:00
commit 459ec2b0a2
22 changed files with 1264 additions and 25 deletions

View File

@ -26,6 +26,8 @@ public class SGameSetting implements BaseConfig {
private int equipNumlimit;
private int worldTalking;
private static SGameSetting gameSetting;
@ -93,4 +95,7 @@ public class SGameSetting implements BaseConfig {
return result;
}
public int getWorldTalking() {
return worldTalking;
}
}

View File

@ -29,4 +29,11 @@ public interface GlobalsDef {
int LEVLE_LOCK = 1; //1工坊熟练度等级解锁
int BLUE_LOCK = 2; //蓝图解锁
//作坊功能子类型
int WORK_BASE_TYPE =1; //基础锻造
int WORK_CREATE_TYPE =2; // 制作装备
//redis 过期时间
int REDIS_OVER_TIME =-1;
}

View File

@ -1,9 +1,12 @@
package com.ljsd.jieling.db.redis;
import com.google.gson.Gson;
import org.springframework.beans.factory.annotation.Autowired;
import com.ljsd.jieling.util.TimeUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.util.CollectionUtils;
import java.util.List;
@ -13,9 +16,13 @@ import java.util.concurrent.TimeUnit;
public class RedisUtil {
private static final Logger LOGGER = LoggerFactory.getLogger(RedisUtil.class);
private Gson gson = new Gson();
private static int MAX_TRY_TIMES = 3; //最大尝试次数,保障获取/存储成功
private static int FAILED_SLEEP = 2; //每次失败最大停顿时间
private RedisUtil() {
}
@ -27,7 +34,6 @@ public class RedisUtil {
private static final RedisUtil instance = new RedisUtil();
}
@Autowired
private StringRedisTemplate redisTemplate;
@ -664,4 +670,53 @@ public class RedisUtil {
}
}
//操作sortset
public void zsetAddAall(String key, Set<ZSetOperations.TypedTuple<String>> items){
for (int i = 0; i < MAX_TRY_TIMES; i++) {
try {
redisTemplate.opsForZSet().add(key,items);
return;
} catch (Exception e) {
TimeUtils.sleep(FAILED_SLEEP);
}
}
}
public Set<String> getZset(String key,double min,double max){
for (int i = 0; i < MAX_TRY_TIMES; i++) {
try {
return redisTemplate.opsForZSet().rangeByScore(key, min, max);
} catch (Exception e) {
TimeUtils.sleep(FAILED_SLEEP);
}
}
return null;
}
public void removeZSetRange(String key,int nums){
for (int i = 0; i < MAX_TRY_TIMES; i++) {
try {
redisTemplate.opsForZSet().removeRange(key,0,nums);
return;
} catch (Exception e) {
TimeUtils.sleep(FAILED_SLEEP);
}
}
}
public long increment(String key) {
for (int i = 0; i < MAX_TRY_TIMES; i++) {
try {
return redisTemplate.opsForValue().increment(key, 1);
} catch (Exception e) {
TimeUtils.sleep(FAILED_SLEEP);
}
}
return -1;
}
}

View File

@ -0,0 +1,24 @@
package com.ljsd.jieling.handler.chat;
import com.ljsd.jieling.handler.BaseHandler;
import com.ljsd.jieling.logic.chat.ChatLogic;
import com.ljsd.jieling.netty.cocdex.PacketNetData;
import com.ljsd.jieling.network.session.ISession;
import com.ljsd.jieling.protocols.ChatProto;
import com.ljsd.jieling.protocols.MessageTypeProto;
import org.springframework.stereotype.Component;
@Component
public class SendChatInfoHandler extends BaseHandler {
@Override
public MessageTypeProto.MessageType getMessageCode() {
return MessageTypeProto.MessageType.SEND_CHAT_INFO_REQUEST;
}
@Override
public void process(ISession iSession, PacketNetData netData) throws Exception {
byte[] bytes = netData.parseClientProtoNetData();
ChatProto.SendChatInfoRequest sendChatInfoRequest = ChatProto.SendChatInfoRequest.parseFrom(bytes);
ChatLogic.getInstance().sendChatInfo(iSession,sendChatInfoRequest);
}
}

View File

@ -0,0 +1,25 @@
package com.ljsd.jieling.handler.chat;
import com.ljsd.jieling.handler.BaseHandler;
import com.ljsd.jieling.logic.chat.ChatLogic;
import com.ljsd.jieling.netty.cocdex.PacketNetData;
import com.ljsd.jieling.network.session.ISession;
import com.ljsd.jieling.protocols.ChatProto;
import com.ljsd.jieling.protocols.MessageTypeProto;
import org.springframework.stereotype.Component;
@Component
public class SwitchWorldChannelHandler extends BaseHandler {
@Override
public MessageTypeProto.MessageType getMessageCode() {
return MessageTypeProto.MessageType.SWITCH_WORLDCHANNEL_REQUEST;
}
@Override
public void process(ISession iSession, PacketNetData netData) throws Exception {
byte[] bytes = netData.parseClientProtoNetData();
ChatProto.SwitchWorldChannelRequest switchWorldChannelRequest = ChatProto.SwitchWorldChannelRequest.parseFrom(bytes);
int channelId = switchWorldChannelRequest.getChannelId();
ChatLogic.getInstance().switchWorldChannel(iSession,channelId);
}
}

View File

@ -0,0 +1,24 @@
package com.ljsd.jieling.handler.workshop;
import com.ljsd.jieling.handler.BaseHandler;
import com.ljsd.jieling.logic.item.WorkShopLogic;
import com.ljsd.jieling.netty.cocdex.PacketNetData;
import com.ljsd.jieling.network.session.ISession;
import com.ljsd.jieling.protocols.MessageTypeProto;
import com.ljsd.jieling.protocols.PlayerInfoProto;
import org.springframework.stereotype.Component;
@Component
public class CookFoodHandler extends BaseHandler {
@Override
public MessageTypeProto.MessageType getMessageCode() {
return MessageTypeProto.MessageType.COOK_FOOD_REQUEST;
}
@Override
public void process(ISession iSession, PacketNetData netData) throws Exception {
byte[] bytes = netData.parseClientProtoNetData();
PlayerInfoProto.CookFoodRequest cookFoodRequest = PlayerInfoProto.CookFoodRequest.parseFrom(bytes);
WorkShopLogic.getInstance().makeFood(iSession,cookFoodRequest.getMaterialIdList(),cookFoodRequest.getNums());
}
}

View File

@ -0,0 +1,37 @@
package com.ljsd.jieling.logic.chat;
import com.ljsd.jieling.config.SGameSetting;
import com.ljsd.jieling.network.session.ISession;
import com.ljsd.jieling.protocols.ChatProto;
import com.ljsd.jieling.protocols.MessageTypeProto;
import com.ljsd.jieling.util.MessageUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ChatLogic {
private static final Logger LOGGER = LoggerFactory.getLogger(ChatLogic.class);
private ChatLogic(){}
public static class Instance {
public final static ChatLogic instance = new ChatLogic();
}
public static ChatLogic getInstance() {
return ChatLogic.Instance.instance;
}
public void switchWorldChannel(ISession iSession, int channelId) throws Exception {
if(channelId> SGameSetting.getGameSetting().getWorldTalking()){
MessageUtil.sendErrorResponse(iSession,0, MessageTypeProto.MessageType.SWITCH_WORLDCHANNEL_RESPONSE_VALUE,"参数错误");
return;
}
//移除自己所在的频道。
}
public void sendChatInfo(ISession iSession, ChatProto.SendChatInfoRequest sendChatInfoRequest) {
}
}

View File

@ -0,0 +1,43 @@
package com.ljsd.jieling.logic.chat;
/**
* Created by Administrator on 2016/3/17.
*/
public class ChatRedisKeyUtil {
private static String separator = "_" ;
private static String worldMessage = "wmsg" ;
private static String unionMessage = "umsg" ;
private static String privateMessage = "pmsg" ;
private static String systemMessage = "smsg" ;
private static String endTailed = "key";
private static String sendMessage = "send";
public static String getWorldMessageKey(){
return worldMessage+separator+endTailed;
}
public static String getUnionMessageKey(int unionId){
return unionMessage+separator+unionId+separator+endTailed;
}
public static String getPrivateMessageKey(String sender, String targetUid) {
return privateMessage + separator + sender + separator + targetUid + separator + endTailed;
}
// 存谁给 targetUid 发送过消息
public static String getPrivateSenderKey(int targetUid) {
return privateMessage + separator + targetUid + separator + endTailed;
}
//暂时不用
public static String getSystemMessageKey(){
return systemMessage+separator+endTailed;
}
// 存sendUid 给谁发送过消息
public static String getTargetSenderKey(int sendUid) {
return sendMessage + separator + sendUid + separator + endTailed;
}
}

View File

@ -0,0 +1,30 @@
package com.ljsd.jieling.logic.chat;
import com.ljsd.jieling.protocols.ChatProto;
public class ChatWorldMessage {
private long messsageId;
private ChatProto.ChatIndication msg;
public ChatWorldMessage(long messsageId, ChatProto.ChatIndication msg) {
this.messsageId = messsageId;
this.msg = msg;
}
public long getMesssageId() {
return messsageId;
}
public void setMesssageId(long messsageId) {
this.messsageId = messsageId;
}
public ChatProto.ChatIndication getMsg() {
return msg;
}
public void setMsg(ChatProto.ChatIndication msg) {
this.msg = msg;
}
}

View File

@ -0,0 +1,111 @@
package com.ljsd.jieling.logic.chat;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.List;
public class LoopQueue<T> {
class Item{
T obj;
long timestamp;
long messgeId;
Item(){
}
void setObj(T obj,long messgeId){
this.obj = obj;
this.timestamp = System.currentTimeMillis();
this.messgeId = messgeId;
}
}
private int writeIndex;
private Item[] queueList;
private int maxlen = 0;
private int getMaxLen =0;
private long diffTime ;
private int initLength =20;
public LoopQueue(int maxlen,int getMaxLen, long diffTime) {
this.maxlen = maxlen;
queueList = (Item[])Array.newInstance(Item.class, maxlen);
for(int i=0; i<maxlen; i++){
queueList[i] = new Item();
}
writeIndex = 0;
this.getMaxLen = getMaxLen;
this.diffTime = diffTime;
}
//添加一个新的元素到循环队列中
public synchronized void push(T obj,long messageId) {
this.queueList[writeIndex%maxlen].setObj(obj,messageId);
writeIndex = writeIndex+1;
}
//diffTime 允许的取到的最小时间差的数据,为-1时则不校验时间差
//getMaxLen 允许取到的最大数量
public synchronized List<T> getObjs(long messageId) {
List<T> objs = new ArrayList<>();
int nextIndex = writeIndex%maxlen-1;
int endIndex = writeIndex%maxlen;
if(nextIndex==-1){
nextIndex = maxlen - 1;
}
boolean needCompareTime = (messageId == 0);
long curTime = System.currentTimeMillis();
while(queueList[nextIndex].obj != null && nextIndex != endIndex){
if(!needCompareTime){
//比较消息id
if(messageId>=queueList[nextIndex].messgeId){
break;
}
}
if (needCompareTime && objs.size() >= initLength ){
break;
}
if(needCompareTime){
if( (curTime - queueList[nextIndex].timestamp) >= diffTime){
break;
}
}
objs.add(0, queueList[nextIndex].obj);
if(objs.size()>maxlen){
break;
}
nextIndex--;
if(nextIndex == -1){
nextIndex = maxlen - 1;
}
}
if(!needCompareTime && objs.size()>getMaxLen){
return objs.subList(0,getMaxLen);
}
return objs;
}
public synchronized String toString() {
StringBuilder builder = new StringBuilder();
for(int i=0; i<maxlen; i++){
builder.append(i);
if(i == writeIndex%maxlen){
builder.append("(writeIndex)");
}
builder.append("->");
if(queueList[i].obj == null){
builder.append("-1");
} else {
builder.append(queueList[i].obj);
}
builder.append(",");
}
return builder.toString();
}
}

View File

@ -0,0 +1,40 @@
package com.ljsd.jieling.logic.chat;
import com.ljsd.jieling.network.session.ISession;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.HashMap;
import java.util.Map;
public class MessageChatCache {
private static final Logger LOGGER = LoggerFactory.getLogger(MessageChatCache.class);
private static final int capuatiy =200;
private static final int msgInterval = 24*3600*1000;
private static final int perCaught= 50;
public static Map<Integer,LoopQueue> worldMsgMap = new HashMap<>();
public static LoopQueue sysMsg = new LoopQueue(capuatiy,perCaught,msgInterval);
public static LoopQueue redEnvelopeMsg = new LoopQueue(capuatiy,perCaught,msgInterval);
//type 0:系统 1世界聊天
public static void getMessage(ISession session){
}
public static void addWordMsg(int channel,ChatWorldMessage wordMessage){
if(!worldMsgMap.containsKey(channel)){
worldMsgMap.put(channel,new LoopQueue(capuatiy, perCaught, msgInterval));
}
worldMsgMap.get(channel).push(wordMessage,wordMessage.getMesssageId());
}
public static void addSystemMsg(ChatWorldMessage sysMessage){
sysMsg.push(sysMessage,sysMessage.getMesssageId());
}
public static void addRedEnvelopeMsg(ChatWorldMessage sysMessage){
redEnvelopeMsg.push(sysMessage,sysMessage.getMesssageId());
}
}

View File

@ -0,0 +1,41 @@
package com.ljsd.jieling.logic.chat;
/**
* Created by Administrator on 2016/2/26.
*/
public enum MessageType {
SystemMsg(0),
WorldMsg(1),
UnionMsg(2),
TeamMsg(3),
PrivateMsg(4);
private final int value;
private MessageType(int value) {
this.value = value;
}
public int getValue() {
return value;
}
public static String findByValueString(int value) {
switch (value) {
case 0:
return "system_msg";
case 1:
return "world_msg";
case 2:
return "union_Msg";
case 3:
return "team_Msg";
case 4:
return "private_msg";
default:
return null;
}
}
}

View File

@ -60,7 +60,7 @@ public class User extends MongoRoot {
this.equipManager.init(this,"equipManager",false);
this.adventureManager.init(this,"adventureManager",false);
this.pokemonManager.init(this,"pokemonManager",false);
// this.workShopController.init(this,"workShopController",false);
this.workShopController.init(this,"workShopController",false);
}
public User(){
@ -73,6 +73,7 @@ public class User extends MongoRoot {
this.equipManager = new EquipManager();
this.adventureManager = new AdventureManager();
this.pokemonManager = new PokemonManager();
this.workShopController = new WorkShopController();
//綁定关系
playerManager.init(this, "playerManager", false);
itemManager.init(this, "itemManager", false);
@ -82,7 +83,7 @@ public class User extends MongoRoot {
missionManager.init(this, "missionManager", false);
adventureManager.init(this,"adventureManager",false);
pokemonManager.init(this,"pokemonManager",false);
// workShopController.init(this,"workShopController",false);
workShopController.init(this,"workShopController",false);
}
public static void init(LjsdMongoTemplate ljsdMongoTemplate) {

View File

@ -55,12 +55,18 @@ public class WorkShopLogic {
if(sMysteryFoodSetting.getNum() != materials.size()){
return;
}
PlayerManager playerInfoManager = user.getPlayerInfoManager();
SWorkShopSetting sWorkShopSetting = SWorkShopSetting.getsWorkShopSettingByLevel(playerInfoManager.getLevel());
WorkShopController workShopController = user.getWorkShopController();
SWorkShopSetting sWorkShopSetting = SWorkShopSetting.getsWorkShopSettingByLevel(workShopController.getCookLevel());
float baseLevelRate = sWorkShopSetting.getAddSuccessRate();
float addRate =0.0f;
Map<Integer, Float> addRateByQuality = sMysteryFoodSetting.getAddRateByQuality();
Set<Integer> cache = new HashSet<>();
for(Integer id : materials){
if(cache.contains(id)){
MessageUtil.sendMessage(session,1, MessageTypeProto.MessageType.COOK_FOOD_RESPONSE_VALUE,null,true);
return;
}
cache.add(id);
Map<Integer, SItem> integerSItemMap = SItem.getsItemMap();
SItem sItem = integerSItemMap.get(id);
if(!addRateByQuality.containsKey(sItem.getQuantity())){
@ -97,13 +103,13 @@ public class WorkShopLogic {
totalFailExp+=failExp;
}
CommonProto.Drop.Builder dropThing = ItemUtil.drop(user, drop);
if(totalFailExp>0){
user.getWorkShopController().addCookExp(totalFailExp);
}
CommonProto.WorkShopBaseInfo workShopBaseInfo = CommonProto.WorkShopBaseInfo.newBuilder().setType(GlobalsDef.COOK_SHOP_TYPE).setLevle(workShopController.getCookLevel()).setExp(workShopController.getCookExp()).build();
PlayerInfoProto.CookFoodResponse build = PlayerInfoProto.CookFoodResponse.newBuilder().setWorkShopBaseInfo(workShopBaseInfo).setDrop(dropThing).build();
MessageUtil.sendMessage(session,1, MessageTypeProto.MessageType.COOK_FOOD_RESPONSE_VALUE,build,true);
}
@ -123,7 +129,7 @@ public class WorkShopLogic {
int uid = session.getUid();
User user = UserManager.getUser(uid);
//基础打造
if(type ==1){
if(type ==GlobalsDef.WORK_BASE_TYPE){
SWorkShopFoundationConfig sWorkShopFoundationConfig = SWorkShopFoundationConfig.getShopFoundationConfigMap().get(functionId);
if(!checkForActiveUseBlue(user,sWorkShopFoundationConfig.getOpenRules())){
MessageUtil.sendErrorResponse(session,0, MessageTypeProto.MessageType.ACTIVITE_WORKSHOP_RESPONSE_VALUE,"参数有误");
@ -132,7 +138,7 @@ public class WorkShopLogic {
}
//装备打造
if(type == 2){
if(type == GlobalsDef.WORK_CREATE_TYPE){
SWorkShopEquipmentConfig sWorkShopEquipmentConfig = SWorkShopEquipmentConfig.getsWorkShopEquipmentConfigMap().get(functionId);
if(!checkForActiveUseBlue(user,sWorkShopEquipmentConfig.getOpenRules())){
MessageUtil.sendErrorResponse(session,0, MessageTypeProto.MessageType.ACTIVITE_WORKSHOP_RESPONSE_VALUE,"参数有误");
@ -171,18 +177,14 @@ public class WorkShopLogic {
List<Integer> workShopList = openBlueStateMap.get(GlobalsDef.WORK_SHOP_TYPE);
List<Integer> cookShopList = openBlueStateMap.get(GlobalsDef.COOK_SHOP_TYPE);
List<CommonProto.WorkShopInfo> workShopInfos = new ArrayList<>(2);
workShopInfos.add( CommonProto.WorkShopInfo.newBuilder().setType(GlobalsDef.WORK_SHOP_TYPE)
.setLevle(workShopLevel)
.setExp(workShopExp)
.addAllId(workShopList == null ? new ArrayList<>(1):workShopList)
.build());
workShopInfos.add( CommonProto.WorkShopInfo.newBuilder().setType(GlobalsDef.COOK_SHOP_TYPE)
.setLevle(cookLevel)
.setExp(cookExp)
.addAllId(cookShopList == null ? new ArrayList<>(1):cookShopList)
.build());
PlayerInfoProto.GetWorkShopInfoResponse build = PlayerInfoProto.GetWorkShopInfoResponse.newBuilder().addAllWorkShopInfo(workShopInfos).build();
List<CommonProto.WorkShopBaseInfo> workShopBaseInfoList = new ArrayList<>(2);
workShopBaseInfoList.add(CommonProto.WorkShopBaseInfo.newBuilder().setType(GlobalsDef.WORK_SHOP_TYPE).setLevle(workShopLevel).setExp(workShopExp).build());
workShopBaseInfoList.add(CommonProto.WorkShopBaseInfo.newBuilder().setType(GlobalsDef.COOK_SHOP_TYPE).setLevle(cookLevel).setExp(cookExp).build());
List<CommonProto.WorkShopUnLockInfo> workShopUnLockInfoList = new ArrayList<>(2);
workShopUnLockInfoList.add(CommonProto.WorkShopUnLockInfo.newBuilder().setType(GlobalsDef.WORK_BASE_TYPE).addAllId(workShopList).build());
workShopUnLockInfoList.add(CommonProto.WorkShopUnLockInfo.newBuilder().setType(GlobalsDef.WORK_CREATE_TYPE).addAllId(cookShopList).build());
PlayerInfoProto.GetWorkShopInfoResponse build = PlayerInfoProto.GetWorkShopInfoResponse.newBuilder().addAllWorkShopBaseInfo(workShopBaseInfoList).addAllWorkShopUnLockInfo(workShopUnLockInfoList).build();
MessageUtil.sendMessage(session,1, MessageTypeProto.MessageType.GET_WORKSHOP_INFO_RESPONSE_VALUE,build,true);
}
@ -292,8 +294,8 @@ public class WorkShopLogic {
equip.rebuildEquip(user.getWorkShopController().getWorkShopLevel());
equipManager.addEquip(equip);
MessageUtil.sendMessage(session,1, MessageTypeProto.MessageType.WORKSHOP_REBUILD_RESPONSE_VALUE,null,true);
PlayerInfoProto.WorkShopRebuildRespoonse build = PlayerInfoProto.WorkShopRebuildRespoonse.newBuilder().setEquip(CBean2Proto.getEquipProto(equip)).build();
MessageUtil.sendMessage(session,1, MessageTypeProto.MessageType.WORKSHOP_REBUILD_RESPONSE_VALUE,build,true);
}

View File

@ -0,0 +1,28 @@
package com.ljsd.jieling.mq;
import com.ljsd.jieling.db.redis.RedisUtil;
import org.slf4j.LoggerFactory;
public class RedisCleanRecordThread extends Thread{
private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(RedisCleanRecordThread.class);
private String topic;
private String type;
private int deleteNums;
public RedisCleanRecordThread(String topic, String type,int deleteNums) {
this.topic = topic;
this.type = type;
this.deleteNums = deleteNums;
}
@Override
public void run() {
try {
// RedisUtil.getInstence().removeZSetRange(type,topic,deleteNums);
}catch (Exception e){
LOGGER.error("e",e);
}
}
}

View File

@ -0,0 +1,26 @@
package com.ljsd.jieling.mq;
public class RedisRecord {
private String key;
private String value;
private String topic;
public RedisRecord(String topic,String key, String value) {
this.topic = topic;
this.key = key;
this.value = value;
}
public String value(){
return value;
}
public String key(){
return key;
}
public String getTopic() {
return topic;
}
}

View File

@ -0,0 +1,70 @@
package com.ljsd.jieling.mq;
import com.google.gson.Gson;
import com.ljsd.jieling.db.redis.RedisUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.ZSetOperations;
import java.util.*;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
public class RedisRecordSendThread extends Thread{
private static final Logger LOGGER = LoggerFactory.getLogger(RedisRecordSendThread.class);
private final BlockingQueue<RedisRecord> redisRecords = new ArrayBlockingQueue<RedisRecord>(500);
private Gson gson = new Gson();
public static RedisRecordSendThread getInstance() {
return Instance.instance;
}
public static class Instance {
public final static RedisRecordSendThread instance = new RedisRecordSendThread();
}
private RedisRecordSendThread(){
}
@Override
public void run() {
while (true){
dowork();
}
}
private void dowork() {
try {
List<RedisRecord> willSend = new ArrayList<>();
redisRecords.drainTo(willSend,10);
if(willSend.isEmpty()){
try {
Thread.sleep(2);
return;
} catch (InterruptedException e) {
e.printStackTrace();
}
}
Set<ZSetOperations.TypedTuple<String>> result = new HashSet<>(willSend.size());
Map<String,Set<ZSetOperations.TypedTuple<String>>> map = new HashMap<>();
for(RedisRecord redisRecord : willSend){
String itemInfo = gson.toJson(redisRecord);
String topic = redisRecord.getTopic();
long increment = RedisUtil.getInstence().increment(topic+"_num");
if(!map.containsKey(topic)){
map.put(topic,new LinkedHashSet<>());
}
map.get(topic).add(new DefaultTypedTuple<String>(itemInfo,(double)increment));
}
for(String topic : map.keySet()){
RedisUtil.getInstence().zsetAddAall(topic,map.get(topic));
}
}catch (Exception e){
LOGGER.error("the exception",e);
}
}
public void addEntry(RedisRecord redisRecord){
redisRecords.add(redisRecord);
}
}

View File

@ -0,0 +1,102 @@
//package com.ljsd.jieling.mq.consumeer;
//
//import com.google.common.util.concurrent.ThreadFactoryBuilder;
//import com.google.gson.Gson;
//import com.jmfy.xianxia.AreaManager;
//import com.jmfy.xianxia.core.GlobalsDef;
//import com.jmfy.xianxia.core.util.RedisUtil;
//import com.jmfy.xianxia.kafka.RedisCleanRecordThread;
//import com.jmfy.xianxia.kafka.RedisConsumeHandler;
//import com.jmfy.xianxia.kafka.RedisRecord;
//import com.jmfy.xianxia.redisProperties.RedisUserKey;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.data.redis.core.StringRedisTemplate;
//
//import java.util.Set;
//import java.util.concurrent.*;
//
//public class RedisChatConsumeHandler extends Thread{
// private static final Logger LOGGER = LoggerFactory.getLogger(RedisConsumeHandler.class);
//
// private long start;
// private StringRedisTemplate stringRedisTemplate;
// private static String topic;
//
// private ExecutorService executors;
//
// private int nums;
//
//
//
//
// private Gson gson = new Gson();
//
// public static RedisChatConsumeHandler getInstance() {
// return Instance.instance;
// }
//
// public static class Instance {
// public final static RedisChatConsumeHandler instance = new RedisChatConsumeHandler();
// }
//
//
// private RedisChatConsumeHandler(){
//
// }
//
// @Override
// public void run() {
// while (true){work();}
// }
//
// public void init(int workerNum){
// topic = "msg_chat_"+ AreaManager.areaId;
// Long msgNum = RedisUtil.getInstence().getObject(RedisUserKey.RedisRecordMQ, topic+"_num", Long.class, GlobalsDef.REDIS_OVER_TIME);
// if(msgNum == null) {
// msgNum = 0L;
// }
//
// start =msgNum+1;
// ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("RedisConsumeHandler-%d").build();
// executors = new ThreadPoolExecutor(workerNum,
// workerNum, 0L, TimeUnit.MILLISECONDS,
// new ArrayBlockingQueue<Runnable>(10000),
// threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());
// this.start();
// }
//
// public void work(){
// try {
// Set<String> result = RedisUtil.getInstence().getZset(RedisUserKey.RedisRecordMQ,topic,start, start + 30);
// if(result == null || result.isEmpty()){
// try {
// Thread.sleep(20);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// return;
// }
// start = result.size() + start;
// for(String key : result){
// RedisRecord redisRecord = gson.fromJson(key, RedisRecord.class);
// if (!redisRecord.value().isEmpty()) {
// nums++;
// executors.submit(new MessageWorker(redisRecord));
// }
// }
// if(nums>100){
// nums=0;
// executors.submit(new RedisCleanRecordThread(topic,RedisUserKey.RedisRecordMQ,50));
// }
// }catch (Exception e){
// LOGGER.error("the exception",e);
// }
//
// }
//
// public static void resubsribe(){
// String changeTopics = "msg_chat_" + AreaManager.areaId;
// topic = changeTopics;
// }
//}

View File

@ -0,0 +1,213 @@
//package com.ljsd.jieling.mq.consumeer;
//
//import com.google.common.util.concurrent.ThreadFactoryBuilder;
//import com.google.gson.Gson;
//import com.ljsd.jieling.core.GlobalsDef;
//import com.ljsd.jieling.db.redis.RedisUtil;
//import com.ljsd.jieling.exception.ErrorCode;
//import com.ljsd.jieling.mq.RedisCleanRecordThread;
//import com.ljsd.jieling.mq.RedisRecord;
//import com.ljsd.jieling.mq.work.PushWorker;
//import com.ljsd.jieling.network.server.ProtocolsManager;
//import org.slf4j.Logger;
//import org.slf4j.LoggerFactory;
//import org.springframework.data.redis.cache.RedisCache;
//import org.springframework.data.redis.core.StringRedisTemplate;
//
//import java.util.ArrayList;
//import java.util.Set;
//import java.util.concurrent.*;
//
//public class RedisConsumeHandler extends Thread{
// private static final Logger LOGGER = LoggerFactory.getLogger(RedisConsumeHandler.class);
//
// private long start;
// private StringRedisTemplate stringRedisTemplate;
// private static String topic;
//
// private ExecutorService executors;
// private int nums;
// private String idipTopic;
// private long idipStart;
// private int idipNums;
//
// private Gson gson = new Gson();
// private ArrayList<DelayMessage> delayMessages = new ArrayList();
// private int curDelayMessageIndex = -1;
//
// public static RedisConsumeHandler getInstance() {
// return Instance.instance;
// }
//
// public static class Instance {
// public final static RedisConsumeHandler instance = new RedisConsumeHandler();
// }
//
// private static class DelayMessage {
// public String topKey;
// public String topValue;
// public long firstDealTime;
// }
//
// private RedisConsumeHandler(){
//
// }
//
// @Override
// public void run() {
// while (true){
// work();
// }
// }
//
// public void init(int workerNum){
// topic = "msg_"+ Server;
// idipTopic = Application.serverId + ":idip";
// Long msgNum = RedisUtil.getInstence().getObject(RedisUserKey.RedisRecordMQ, topic+"_num", Long.class, GlobalsDef.REDIS_OVER_TIME);
// if(msgNum == null){
// msgNum =-1L;
// }
// start =msgNum+1;
//
// Long idipMsgNum = RedisUtil.getInstence().getObject(RedisUserKey.RedisRecordIdipMQ, idipTopic+"_num", Long.class, GlobalsDef.REDIS_OVER_TIME);
// if(idipMsgNum == null){
// idipMsgNum =-1L;
// }
// idipStart =idipMsgNum+1;
// ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("RedisConsumeHandler-%d").build();
// executors = new ThreadPoolExecutor(workerNum,
// workerNum, 0L, TimeUnit.MILLISECONDS,
// new ArrayBlockingQueue<Runnable>(10000),
// threadFactory, new ThreadPoolExecutor.CallerRunsPolicy());
// start();
// }
//
//
// //优先处理idip消息
// public boolean processIdIpMsg(){
// Set<String> result = RedisUtil.getInstence().getZset(RedisUserKey.RedisRecordIdipMQ,idipTopic,idipStart, idipStart + 30);
// if(result == null || result.isEmpty()){
// return false;
// }
// idipStart = result.size() + idipStart;
// for(String key : result){
// RedisRecord redisRecord = gson.fromJson(key, RedisRecord.class);
// if (!redisRecord.value().isEmpty()) {
// idipNums++;
// Runnable worker = getWorkerByTopic(redisRecord);
// if (worker != null) {
// String[] split = redisRecord.value().split(":");
// if (RedisUserKey.Kick_Old_Key.equals(split[0])) {
// KafkaConsumerHandler.DelayMessage delayMessage = new KafkaConsumerHandler.DelayMessage();
// delayMessage.topKey = split[0];
// delayMessage.topValue = redisRecord.value();
// delayMessage.firstDealTime = System.currentTimeMillis() / 1000;
// delayMessages.add(delayMessage);
// }
// executors.submit(worker);
// }
// }
// }
// if(idipNums>30){
// idipNums=0;
// executors.submit(new RedisCleanRecordThread(topic,RedisUserKey.RedisRecordIdipMQ,20));
// }
// return true;
// }
//
// private void processKickMsg(){
// if (delayMessages.size() != 0) {
// for (int i = 0; i < delayMessages.size(); i++) {
// KafkaConsumerHandler.DelayMessage delayMessage = delayMessages.get(i);
// String topValue = delayMessage.topValue;
// String[] split = topValue.split(":");
// if (System.currentTimeMillis() / 1000 > (delayMessage.firstDealTime + Integer.parseInt(split[2]))) {
// curDelayMessageIndex = i;
// kickOldUser(Integer.parseInt(split[1]), Integer.parseInt(split[3]));
// delayMessages.remove(curDelayMessageIndex);
// }
// }
// }
// }
//
// public void work(){
// try {
// processKickMsg();
// if(!processIdIpMsg()){
// Set<String> result = RedisUtil.getInstence().getZset(RedisUserKey.RedisRecordMQ,topic,start, start + 30);
// if(result == null || result.isEmpty()){
// try {
// Thread.sleep(200);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// return;
// }
// start = result.size() + start;
// for(String key : result){
// RedisRecord redisRecord = gson.fromJson(key, RedisRecord.class);
// if (!redisRecord.value().isEmpty()) {
// Runnable worker = getWorkerByTopic(redisRecord);
// if (worker != null) {
// nums++;
// executors.submit(worker);
// }
// }
// }
// if(nums>100){
// nums=0;
// executors.submit(new RedisCleanRecordThread(topic,RedisUserKey.RedisRecordMQ,50));
// }
// }
//
// }catch (Exception e){
// LOGGER.error("the exception",e);
// }
//
// }
//
// private Runnable getWorkerByTopic(RedisRecord record) {
// if (record.key().startsWith("tlog")) {
// return new TLogWorker(record);
// }
// if (record.key().startsWith("push")) {
// return new PushWorker(record);
// }
//// else if (topic.startsWith("")){
//// return null;
//// }
//
// return null;
// }
//
// public static void resubsribe(){
// String changeTopics = "msg_" + AreaManager.areaId;
// topic = changeTopics;
// }
//
// private void kickOldUser(int uid, int type) {
// //type 0:全服玩家踢下线 1:单服玩家踢下线
// LOGGER.info("idipKickOldUser-->uid={},type={}", uid, type);
// if (type == 0) {
// kickOldAllUser();
// } else {
// PlayerLogic.getInstance().kickOldOneUser(uid);
// }
// }
//
// private void kickOldAllUser() {
// Set<Integer> onlineUserId = OnlineUserLogics.getOnlineIdSet();
// LOGGER.info("kickOldAllUser-->onlineNum={}", onlineUserId.size());
// for (Integer userId : onlineUserId) {
// if (OnlineUserLogics.isOnline(userId)) {
// ProtocolsManager.getInstance().kickOldUser(userId, "",
// ErrorCode.IDIP_KIC_OLD, "IDIP_KIC_OLD!!!", 0);
// //清除缓存
// ProtocolsManager.getInstance().cleanResponseCache(userId, true);
// RedisCache.cleanCache(userId);
// }
// }
// }
//
//
//}

View File

@ -0,0 +1,51 @@
package com.ljsd.jieling.mq.work;
import com.ljsd.jieling.mq.RedisRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class ChatMsgWorker implements Runnable {
private static final Logger LOGGER = LoggerFactory.getLogger(ChatMsgWorker.class);
//private ConsumerRecord<String, String> consumerRecord;
private RedisRecord consumerRecord;
public ChatMsgWorker(RedisRecord consumerRecord) {
this.consumerRecord = consumerRecord;
}
@Override
public void run() {
try{
/* if (consumerRecord.key().equals(MyKafkaConstance.PRIVATE_MSG)) {
ChatSendUtils.getInstance().sendToFrontendPrivate(consumerRecord.value());
LOGGER.debug("run->msg={};consume private", consumerRecord.value());
} else if (consumerRecord.key().equals(MyKafkaConstance.SYSTEM_MSG)) {
ChatSendUtils.sendToFrontendSystem(consumerRecord.value());
LOGGER.debug("run->msg={};consume system", consumerRecord.value());
} else if (consumerRecord.key().equals(MyKafkaConstance.WORLD_MSG)) {
LOGGER.debug("run->msg={};receive world", consumerRecord.value());
ChatSendUtils.sendToFrontendWorld(consumerRecord.value());
} else if (consumerRecord.key().equals(MyKafkaConstance.UNION_MSG)) {
ChatSendUtils.sendToFrontendFamily(consumerRecord.value());
LOGGER.debug("run->msg={};consume union", consumerRecord.value());
} else if (consumerRecord.key().equals(MyKafkaConstance.UNION_SYSTEM_MSG)) {
ChatSendUtils.sendToFrontendFamilySystem(consumerRecord.value());
LOGGER.debug("run->msg={};consume union system", consumerRecord.value());
}else if (consumerRecord.key().equals(MyKafkaConstance.TEAM_MSG)) {
ChatSendUtils.sendToFrontendTeam(consumerRecord.value());
LOGGER.debug("run->msg={};consume team ", consumerRecord.value());
}*/
}catch (Exception e){
LOGGER.error("run->msg={}", e.getMessage(), e);
}
}
}

View File

@ -0,0 +1,59 @@
package com.ljsd.jieling.mq.work;
import com.ljsd.jieling.mq.RedisRecord;
import com.ljsd.jieling.protocols.ChatProto;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class PushWorker implements Runnable {
private static final Logger LOGGER = LoggerFactory.getLogger(PushWorker.class);
private RedisRecord consumerRecord;
public PushWorker(RedisRecord consumerRecord) {
this.consumerRecord = consumerRecord;
}
@Override
public void run() {
try {
Thread.currentThread().setName("PushWorker");
String[] valueArray = consumerRecord.value().split(":");
String key = valueArray[0];
LOGGER.info("PushWorker key={},valueArray={}", key, consumerRecord.value());
} catch (Exception e) {
LOGGER.error("run->msg={}", e.getMessage(), e);
}
}
private ChatProto.ChatIndication getChatIndicationProto(int serverId, String valueArray) {
String[] split = valueArray.split("\\|");
ChatProto.ChatIndication chatIndication = ChatProto.ChatIndication.newBuilder()
.setSenderId(1001)
.setSenderServerId(serverId)
.setSenderName("IDIP")
.setSenderlevel(100)
.setSenderimg("i_h3_41")
.setSendervip(10)
.setTimes(Integer.parseInt(split[0]))
.setMsgType(0)
.setMsg(split[2])
.setIsFamilySystem(2)
.setEndTime(Integer.parseInt(split[1]))
.setPriorityLevel(Integer.parseInt(split[3]))
.setFrequency(Integer.parseInt(split[4]))
.setIsSystem(Integer.parseInt(split[6]))
.setClientVersion(split[7])
.build();
return chatIndication;
}
}

View File

@ -0,0 +1,245 @@
package com.ljsd.jieling.util;
import com.ljsd.jieling.core.GlobalsDef;
import com.ljsd.jieling.logic.OnlineUserManager;
import com.ljsd.jieling.logic.chat.ChatRedisKeyUtil;
import com.ljsd.jieling.logic.chat.MessageType;
import com.ljsd.jieling.netty.NettyGameSession;
import com.ljsd.jieling.protocols.ChatProto;
import io.netty.channel.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.StringRedisTemplate;
import java.io.UnsupportedEncodingException;
import java.util.*;
import java.util.concurrent.TimeUnit;
/**
* Created by Administrator on 2016/1/26.
*/
public class ChatSendUtils {
private static final Logger LOGGER = LoggerFactory.getLogger(ChatSendUtils.class);
private static ChatSendUtils instance;
public static ChatSendUtils getInstance(){
if (instance == null) {
instance = new ChatSendUtils();
}
return instance;
}
private StringRedisTemplate redisTemplate;
public static void sendToFrontendWorld(String msg) {
if (msg == null || msg.isEmpty()) {
return;
}
String[] msgArray = msg.split("#");
ChatProto.ChatIndication chatIndication = getChatProto(msgArray);
// MessageChatCache.addWordMsg(new ChatWorldMessage(Long.parseLong(msgArray[16]),Long.parseLong(msgArray[17]),chatIndication));
// MessageChatCache.addWordMsg(new ChatWorldMessage(Long.parseLong(msgArray[16]),chatIndication));
}
public void sendToFrontendPrivate(String msg) {
if (msg == null || msg.isEmpty()) {
return;
}
String[] msgArray = msg.split("#");
ChatProto.ChatIndication chatIndication = getChatProto(msgArray);
NettyGameSession mySession = OnlineUserManager.getSessionByUid(Integer.parseInt(msgArray[0]));
NettyGameSession targetSession = OnlineUserManager.getSessionByUid(Integer.parseInt(msgArray[9]));
if (mySession != null) {
sendPrivateMsgIndication(mySession.getCtx().channel(), chatIndication);
}
if (targetSession != null) {
sendPrivateMsgIndication(targetSession.getCtx().channel(), chatIndication);
}
persistLeaveMessage(msgArray[0],msgArray[9],msg);
// else {
// persistLeaveMessage(msgArray[0],msgArray[9],msg);
// persistLeaveMessage(msgArray[9],msgArray[0],msg);
// }
}
public static void sendToFrontendFamily(String msg) {
}
private void persistLeaveMessage(String myId, String targetId, String msg){
String sendersKey = ChatRedisKeyUtil.getPrivateSenderKey(Integer.parseInt(targetId));
String targetSenderKey = ChatRedisKeyUtil.getTargetSenderKey(Integer.parseInt(myId));
List<String> senderList = redisTemplate.opsForList().range(sendersKey, 0, -1);
List<String> rangeList = redisTemplate.opsForList().range(targetSenderKey, 0, -1);
addChatToRedis(myId, sendersKey, senderList);
addChatToRedis(targetId, targetSenderKey, rangeList);
String key = ChatRedisKeyUtil.getPrivateMessageKey(myId, targetId);
List<String> msgList = redisTemplate.opsForList().range(key, 0, -1);
if (msgList.size() > 10) {
redisTemplate.opsForList().rightPop(key);
}
redisTemplate.opsForList().leftPush(key, msg);
}
private void addChatToRedis(String targetId, String targetSenderKey, List<String> rangeList) {
if (!rangeList.isEmpty()){
String ranger = rangeList.get(0);
if (!ranger.contains(targetId)){
redisTemplate.opsForList().rightPop(targetSenderKey);
redisTemplate.opsForList().leftPush(targetSenderKey, ranger + "#" + targetId);
}
}else{
redisTemplate.opsForList().leftPush(targetSenderKey, targetId);
}
if (targetSenderKey.contains("send")){
redisTemplate.expire(targetSenderKey, GlobalsDef.REDIS_OVER_TIME,TimeUnit.SECONDS);
}
}
public static void sendToFrontendSystem(String msg) {
if (msg == null || msg.isEmpty()) {
return;
}
String[] msgArray = msg.split("\\|");
ChatProto.ChatIndication chatIndication = getSystemChatProto(msgArray);
// MessageChatCache.addSystemMsg(new ChatWorldMessage(Long.parseLong(msgArray[14]),Long.parseLong(msgArray[15]),chatIndication));
// MessageChatCache.addSystemMsg(new ChatWorldMessage(Long.parseLong(msgArray[14]),chatIndication));
}
private static void sendPrivateMsgIndication(Channel targetChannel, ChatProto.ChatIndication chatIndication) {
targetChannel.writeAndFlush(chatIndication);
}
public List<ChatProto.ChatIndication> getOffLineChatInfo(int userId, int serverId, MessageType messageType) throws UnsupportedEncodingException {
List<ChatProto.ChatIndication> chatIndications = new ArrayList<>();
List<String> msgs = null;
if (messageType == MessageType.PrivateMsg) {
String sendersKey = ChatRedisKeyUtil.getPrivateSenderKey(userId);
List<String> senderList = redisTemplate.opsForList().range(sendersKey, 0, -1);
if (senderList.isEmpty()){
return chatIndications;
}
String sender = senderList.get(0);
String[] senderArray = sender.split("#");
for (String suid : senderArray){
String key = ChatRedisKeyUtil.getPrivateMessageKey(suid, userId+"");
List<String> messageByte = redisTemplate.opsForList().range(key,0,-1);
if (messageByte==null || messageByte.isEmpty()){
continue;
}
if (msgs == null) {
msgs = new ArrayList<>();
}
Collections.sort(messageByte);
msgs.addAll(messageByte);
// delete
redisTemplate.delete(key);
}
// delete senders
redisTemplate.delete(sendersKey);
} else if (messageType == MessageType.WorldMsg) {
String key = ChatRedisKeyUtil.getWorldMessageKey();
msgs = redisTemplate.opsForList().range(key, 0, -1);
} else if (messageType == MessageType.SystemMsg) {
String key = ChatRedisKeyUtil.getSystemMessageKey();
msgs = redisTemplate.opsForList().range(key, 0, -1);
}else if (messageType == MessageType.TeamMsg){
}
if (msgs==null || msgs.isEmpty()){
return chatIndications;
}
for (String msg : msgs) {
String[] msgArray = msg.split("#");
ChatProto.ChatIndication chatIndication = getChatProto(msgArray);
chatIndications.add(chatIndication);
}
return chatIndications;
}
private static ChatProto.ChatIndication getChatProto(String[] msgArray) {
int msgType = Integer.parseInt(msgArray[7]);
ChatProto.ChatIndication chatIndication = ChatProto.ChatIndication.newBuilder()
.setSenderId(Integer.parseInt(msgArray[0]))
.setSenderServerId(Integer.parseInt(msgArray[1]))
.setSenderName(msgArray[2])
.setSenderlevel(Integer.parseInt(msgArray[3]))
.setSenderimg(msgArray[4])
.setSendervip(Integer.parseInt(msgArray[5]))
.setTimes(Long.parseLong(msgArray[6]))
.setMsgType(Integer.parseInt(msgArray[7]))
.setMsg(msgArray[8])
.setTargetId(Integer.parseInt(msgArray[9]))
.setVoiceMsg(msgArray[11])
.setVoiceTime(Integer.parseInt(msgArray[12]))
.setUnionID(msgArray[13])
.setFrame(Integer.parseInt(msgArray[14]))
.setHead(Integer.parseInt(msgArray[15]))
.setArenaIcon(msgArray[18])
.setArenaRank(Integer.parseInt(msgArray[19]))
.setMsgChildType(Integer.parseInt(msgArray[20]))
.build();
if( msgType == 0 || msgType == 1){
chatIndication = chatIndication.toBuilder().setMessageId(Long.parseLong(msgArray[16])).build();
}
/* if(chatIndication.getMsgChildType() == GlobalsDef.RED_MSG){
ChatProto.ChatIndication redChatbuild = ChatProto.ChatIndication.newBuilder().mergeFrom(chatIndication).setMsgType(9).build();
MessageChatCache.addRedEnvelopeMsg(new ChatWorldMessage(Long.parseLong(msgArray[16]),redChatbuild));
}*/
return chatIndication;
}
private static ChatProto.ChatIndication getGuildSysChatProto(String msg, String guildId) {
ChatProto.ChatIndication chatIndication = ChatProto.ChatIndication.newBuilder()
.setSenderId(0)
.setSenderServerId(0)
.setSenderName("")
.setSenderlevel(0)
.setSenderimg("")
.setSendervip(0)
.setTimes(0)
.setMsgType(2)
.setMsg(msg)
.setTargetId(0)
.setVoiceMsg("")
.setVoiceTime(0)
.setUnionID(guildId)
.setFrame(0)
.setIsFamilySystem(1)
.build();
return chatIndication;
}
private static ChatProto.ChatIndication getSystemChatProto(String[] msgArray) {
ChatProto.ChatIndication chatIndication = ChatProto.ChatIndication.newBuilder()
.setSenderId(Integer.parseInt(msgArray[0]))
.setSenderServerId(Integer.parseInt(msgArray[1]))
.setSenderName(msgArray[2])
.setSenderlevel(Integer.parseInt(msgArray[3]))
.setSenderimg(msgArray[4])
.setSendervip(Integer.parseInt(msgArray[5]))
.setTimes(Long.parseLong(msgArray[6]))
.setMsgType(Integer.parseInt(msgArray[7]))
.setMsg(msgArray[8])
.setVoiceMsg(msgArray[11])
.setVoiceTime(Integer.parseInt(msgArray[12]))
.setPlayCount(Integer.parseInt(msgArray[13]))
.setMessageId(Long.parseLong((msgArray[14])))
.build();
return chatIndication;
}
}