17 changed files with 407 additions and 339 deletions
@ -1,109 +1,108 @@ |
|||||
//package com.ccsens.game.mq;
|
package com.ccsens.game.mq; |
||||
//
|
|
||||
//import cn.hutool.core.collection.CollectionUtil;
|
import cn.hutool.core.collection.CollectionUtil; |
||||
//import cn.hutool.core.util.StrUtil;
|
import cn.hutool.core.util.StrUtil; |
||||
//import com.alibaba.fastjson.JSON;
|
import com.alibaba.fastjson.JSON; |
||||
//import com.alibaba.fastjson.JSONObject;
|
import com.alibaba.fastjson.JSONObject; |
||||
//import com.ccsens.game.bean.dto.ClientDto;
|
import com.ccsens.game.bean.dto.ClientDto; |
||||
//import com.ccsens.game.bean.dto.message.GameMessageCountOut;
|
import com.ccsens.game.bean.dto.message.GameMessageCountOut; |
||||
//import com.ccsens.game.bean.dto.message.GameMessageWithChangeStatusOut;
|
import com.ccsens.game.util.GameConstant; |
||||
//import com.ccsens.game.netty.ChannelManager;
|
import com.ccsens.util.RedisUtil; |
||||
//import com.ccsens.game.util.GameConstant;
|
import com.ccsens.util.bean.message.common.InMessage; |
||||
//import com.ccsens.util.JacksonUtil;
|
import com.ccsens.util.bean.message.common.MessageConstant; |
||||
//import com.ccsens.util.RedisUtil;
|
import com.ccsens.util.bean.message.common.OutMessage; |
||||
//import com.ccsens.util.bean.message.common.InMessage;
|
import com.ccsens.util.bean.message.common.OutMessageSet; |
||||
//import com.ccsens.util.bean.message.common.OutMessage;
|
import com.ccsens.util.config.RabbitMQConfig; |
||||
//import com.ccsens.util.bean.message.common.OutMessageSet;
|
import lombok.extern.slf4j.Slf4j; |
||||
//import com.ccsens.util.config.RabbitMQConfig;
|
import org.springframework.amqp.core.AmqpTemplate; |
||||
//import lombok.extern.slf4j.Slf4j;
|
import org.springframework.amqp.rabbit.annotation.RabbitHandler; |
||||
//import org.springframework.amqp.core.AmqpTemplate;
|
import org.springframework.amqp.rabbit.annotation.RabbitListener; |
||||
//import org.springframework.amqp.rabbit.annotation.RabbitHandler;
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
import org.springframework.data.redis.core.ZSetOperations; |
||||
//import org.springframework.beans.factory.annotation.Autowired;
|
import org.springframework.stereotype.Component; |
||||
//import org.springframework.data.redis.core.ZSetOperations;
|
|
||||
//import org.springframework.stereotype.Component;
|
import java.io.IOException; |
||||
//
|
import java.util.ArrayList; |
||||
//import java.io.IOException;
|
import java.util.HashSet; |
||||
//import java.util.ArrayList;
|
import java.util.List; |
||||
//import java.util.HashSet;
|
import java.util.Set; |
||||
//import java.util.List;
|
|
||||
//import java.util.Set;
|
/** |
||||
//
|
* @description: |
||||
///**
|
* @author: whj |
||||
// * @description:
|
* @time: 2020/6/5 14:30 |
||||
// * @author: whj
|
*/ |
||||
// * @time: 2020/6/5 14:30
|
@Slf4j |
||||
// */
|
@Component |
||||
//@Slf4j
|
@RabbitListener(queues = RabbitMQConfig.GAME_SCORE) |
||||
//@Component
|
public class GameScoreListener { |
||||
//@RabbitListener(queues = RabbitMQConfig.GAME_SCORE)
|
|
||||
//public class GameScoreListener {
|
@Autowired |
||||
//
|
private RedisUtil redisUtil; |
||||
// @Autowired
|
@Autowired |
||||
// private RedisUtil redisUtil;
|
private AmqpTemplate rabbitTemplate; |
||||
// @Autowired
|
|
||||
// private AmqpTemplate rabbitTemplate;
|
@RabbitHandler |
||||
//
|
public void process(String messageJson) throws IOException { |
||||
// @RabbitHandler
|
log.info("计算分数 {}",messageJson); |
||||
// public void process(String messageJson) throws IOException {
|
OutMessageSet outMessageSet = JSONObject.parseObject(messageJson, OutMessageSet.class); |
||||
// log.info("计算分数 {}",messageJson);
|
Set<OutMessage> messageSet = outMessageSet.getMessageSet(); |
||||
// OutMessageSet outMessageSet = JSONObject.parseObject(messageJson, OutMessageSet.class);
|
if (CollectionUtil.isEmpty(messageSet)) { |
||||
// Set<OutMessage> messageSet = outMessageSet.getMessageSet();
|
return; |
||||
// if (CollectionUtil.isEmpty(messageSet)) {
|
} |
||||
// return;
|
messageSet.forEach(outMessage -> { |
||||
// }
|
|
||||
// messageSet.forEach(outMessage -> {
|
InMessage.To to = JSONObject.parseObject(outMessage.getFrom(), InMessage.To.class); |
||||
//
|
Long userId = to.getId(); |
||||
// InMessage.To to = JSONObject.parseObject(outMessage.getFrom(), InMessage.To.class);
|
GameDto.GameMsg gameMsg = JSONObject.parseObject(outMessage.getData(), GameDto.GameMsg.class); |
||||
// Long userId = to.getId();
|
Long recordId = gameMsg.getRecordId(); |
||||
// GameDto.GameMsg gameMsg = JSONObject.parseObject(outMessage.getAuthMessage(), GameDto.GameMsg.class);
|
log.info("统计分数：{},{}",userId, recordId); |
||||
// Long recordId = gameMsg.getRecordId();
|
GameMessageCountOut gameMessageCountOut = clientAddTimes(userId,recordId); |
||||
// log.info("统计分数：{},{}",userId, recordId);
|
|
||||
// GameMessageCountOut gameMessageCountOut = clientAddTimes(userId,recordId);
|
// 发送给消息系统
|
||||
//
|
InMessage inMessage = new InMessage(); |
||||
// // 发送给消息系统
|
Set<String> tos = new HashSet<>(); |
||||
// InMessage inMessage = new InMessage();
|
tos.add(JSONObject.toJSONString(new InMessage.To(userId))); |
||||
// Set<String> tos = new HashSet<>();
|
inMessage.setToDomain(MessageConstant.DomainType.User); |
||||
// tos.add(JSONObject.toJSONString(new InMessage.To(userId)));
|
inMessage.setTos(tos); |
||||
// inMessage.setTos(tos);
|
inMessage.setData(JSONObject.toJSONString(gameMessageCountOut.getData())); |
||||
// inMessage.setData(JSONObject.toJSONString(gameMessageCountOut));
|
rabbitTemplate.convertAndSend(RabbitMQConfig.MESSAGE_QUEUE_NAME, inMessage); |
||||
// rabbitTemplate.convertAndSend(RabbitMQConfig.GAME_SCORE, inMessage);
|
}); |
||||
// });
|
|
||||
//
|
} |
||||
// }
|
|
||||
//
|
public GameMessageCountOut clientAddTimes(Long userId, Long recordId) { |
||||
// public GameMessageCountOut clientAddTimes(Long userId, Long recordId) {
|
GameMessageCountOut gameMessageCountOut = new GameMessageCountOut(); |
||||
// GameMessageCountOut gameMessageCountOut = new GameMessageCountOut();
|
log.info("userId:{}", userId); |
||||
// log.info("userId:{}", userId);
|
if (userId == null || recordId == null) { |
||||
// if (userId == null || recordId == null) {
|
return gameMessageCountOut; |
||||
// return gameMessageCountOut;
|
} |
||||
// }
|
|
||||
//
|
String gameUserKey = GameConstant.generateGameKey(recordId); |
||||
// String gameUserKey = GameConstant.generateGameKey(recordId);
|
Set<ZSetOperations.TypedTuple<Object>> typedTuples = redisUtil.zsRevGetWithScore(gameUserKey, 0, -1); |
||||
// Set<ZSetOperations.TypedTuple<Object>> typedTuples = redisUtil.zsRevGetWithScore(gameUserKey, 0, -1);
|
|
||||
//
|
if (CollectionUtil.isEmpty(typedTuples)) { |
||||
// if (CollectionUtil.isEmpty(typedTuples)) {
|
return gameMessageCountOut; |
||||
// return gameMessageCountOut;
|
} |
||||
// }
|
for(ZSetOperations.TypedTuple<Object> type : typedTuples) { |
||||
// for(ZSetOperations.TypedTuple<Object> type : typedTuples) {
|
ClientDto.RedisUser user = JSON.parseObject((String)type.getValue(), ClientDto.RedisUser.class); |
||||
// ClientDto.RedisUser user = JSON.parseObject((String)type.getValue(), ClientDto.RedisUser.class);
|
if (user.getUserId().longValue() == userId.longValue()) { |
||||
// if (user.getUserId().longValue() == userId.longValue()) {
|
int score = type.getScore().intValue(); |
||||
// int score = type.getScore().intValue();
|
String userStatus = GameConstant.generateGameStatusKey(recordId); |
||||
// String userStatus = GameConstant.generateGameStatusKey(recordId);
|
String gameStausObj = (String)redisUtil.get(userStatus); |
||||
// String gameStausObj = (String)redisUtil.get(userStatus);
|
if (StrUtil.isBlank(gameStausObj) || gameStausObj.equals(String.valueOf(GameConstant.GAME_COMPLETED))){ |
||||
// if (StrUtil.isBlank(gameStausObj) || gameStausObj.equals(String.valueOf(GameConstant.GAME_COMPLETED))){
|
gameMessageCountOut = new GameMessageCountOut(score/100, score); |
||||
// gameMessageCountOut = new GameMessageCountOut(score/100, score);
|
return gameMessageCountOut; |
||||
// return gameMessageCountOut;
|
} |
||||
// }
|
score += 100; |
||||
// score += 100;
|
redisUtil.zsSet(gameUserKey, JSON.toJSONString(user), score); |
||||
// redisUtil.zsSet(gameUserKey, JSON.toJSONString(user), score);
|
gameMessageCountOut.getData().setTotalScore(score); |
||||
// gameMessageCountOut.getData().setTotalScore(score);
|
gameMessageCountOut.getData().setTotalTimes(score/100); |
||||
// gameMessageCountOut.getData().setTotalTimes(score/100);
|
return gameMessageCountOut; |
||||
// return gameMessageCountOut;
|
} |
||||
// }
|
}; |
||||
// };
|
|
||||
//
|
return gameMessageCountOut; |
||||
// return gameMessageCountOut;
|
} |
||||
// }
|
|
||||
//
|
} |
||||
//}
|
|
||||
|
|||||
@ -1,29 +1,29 @@ |
|||||
package com.ccsens.tall.rabbitMQ; |
//package com.ccsens.tall.rabbitMQ;
|
||||
|
//
|
||||
|
//
|
||||
import com.ccsens.util.config.RabbitMQConfig; |
//import com.ccsens.util.config.RabbitMQConfig;
|
||||
import org.slf4j.Logger; |
//import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory; |
//import org.slf4j.LoggerFactory;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitHandler; |
//import org.springframework.amqp.rabbit.annotation.RabbitHandler;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitListener; |
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||
import org.springframework.stereotype.Component; |
//import org.springframework.stereotype.Component;
|
||||
|
//
|
||||
|
//
|
||||
@Component |
//@Component
|
||||
@RabbitListener(queues = RabbitMQConfig.TALL_MESSAGE_2) |
//@RabbitListener(queues = RabbitMQConfig.TALL_MESSAGE_2)
|
||||
public class RabbitController { |
//public class RabbitController {
|
||||
private Logger logger = LoggerFactory.getLogger(RabbitController.class); |
// private Logger logger = LoggerFactory.getLogger(RabbitController.class);
|
||||
|
//
|
||||
@RabbitHandler |
// @RabbitHandler
|
||||
public void process(String messageJson) { |
// public void process(String messageJson) {
|
||||
logger.info("Rabbit Received: {}",messageJson); |
// logger.info("Rabbit Received: {}",messageJson);
|
||||
try { |
// try {
|
||||
System.out.println("Rabbit Received: " + messageJson); |
// System.out.println("Rabbit Received: " + messageJson);
|
||||
|
//
|
||||
}catch (Exception e){ |
// }catch (Exception e){
|
||||
e.printStackTrace(); |
// e.printStackTrace();
|
||||
} |
// }
|
||||
|
//
|
||||
} |
// }
|
||||
|
//
|
||||
} |
//}
|
||||
|
|||||
@ -1,4 +1,4 @@ |
|||||
spring: |
spring: |
||||
profiles: |
profiles: |
||||
active: dev |
active: test |
||||
include: util-dev,common |
include: util-test,common |
||||
|
|||||
@ -1,33 +1,33 @@ |
|||||
package com.ccsens.util.mq; |
//package com.ccsens.util.mq;
|
||||
|
//
|
||||
import com.ccsens.util.config.RabbitMQConfig; |
//import com.ccsens.util.config.RabbitMQConfig;
|
||||
import com.rabbitmq.client.Channel; |
//import com.rabbitmq.client.Channel;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitHandler; |
//import org.springframework.amqp.rabbit.annotation.RabbitHandler;
|
||||
import org.springframework.amqp.rabbit.annotation.RabbitListener; |
//import org.springframework.amqp.rabbit.annotation.RabbitListener;
|
||||
import org.springframework.amqp.support.AmqpHeaders; |
//import org.springframework.amqp.support.AmqpHeaders;
|
||||
import org.springframework.messaging.handler.annotation.Headers; |
//import org.springframework.messaging.handler.annotation.Headers;
|
||||
import org.springframework.messaging.handler.annotation.Payload; |
//import org.springframework.messaging.handler.annotation.Payload;
|
||||
import org.springframework.stereotype.Component; |
//import org.springframework.stereotype.Component;
|
||||
|
//
|
||||
import java.io.IOException; |
//import java.io.IOException;
|
||||
import java.util.Date; |
//import java.util.Date;
|
||||
import java.util.Map; |
//import java.util.Map;
|
||||
|
//
|
||||
/** |
///**
|
||||
* @description: |
// * @description:
|
||||
* @author: wuHuiJuan |
// * @author: wuHuiJuan
|
||||
* @create: 2019/12/27 15:40 |
// * @create: 2019/12/27 15:40
|
||||
*/ |
// */
|
||||
@Component |
//@Component
|
||||
public class DelayConsumer { |
//public class DelayConsumer {
|
||||
@RabbitListener(queues = RabbitMQConfig.GAME_STATUS_DELAY_SEND_QUEUE) |
// @RabbitListener(queues = RabbitMQConfig.GAME_STATUS_DELAY_SEND_QUEUE)
|
||||
@RabbitHandler |
// @RabbitHandler
|
||||
public void receive(@Payload Object msg, @Headers Map<String, Object> headers, Channel channel ) throws IOException { |
// public void receive(@Payload Object msg, @Headers Map<String, Object> headers, Channel channel ) throws IOException {
|
||||
System.out.println("接收到的消息:"+msg +"||" + new Date()); |
// System.out.println("接收到的消息:"+msg +"||" + new Date());
|
||||
|
//
|
||||
//ACK 手工签收，通知rabbitMQ，消费端消费成功
|
// //ACK 手工签收，通知rabbitMQ，消费端消费成功
|
||||
Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG); |
// Long deliveryTag = (Long)headers.get(AmqpHeaders.DELIVERY_TAG);
|
||||
channel.basicAck(deliveryTag,false); |
// channel.basicAck(deliveryTag,false);
|
||||
} |
// }
|
||||
|
//
|
||||
} |
//}
|
||||
|
|||||
@ -1,54 +1,54 @@ |
|||||
package com.ccsens.util.mq; |
//package com.ccsens.util.mq;
|
||||
|
//
|
||||
import com.ccsens.util.config.RabbitMQConfig; |
//import com.ccsens.util.config.RabbitMQConfig;
|
||||
import lombok.extern.slf4j.Slf4j; |
//import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.amqp.rabbit.connection.CorrelationData; |
//import org.springframework.amqp.rabbit.connection.CorrelationData;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate; |
//import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.beans.factory.annotation.Autowired; |
//import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component; |
//import org.springframework.stereotype.Component;
|
||||
|
//
|
||||
/** |
///**
|
||||
* @description: |
// * @description:
|
||||
* @author: wuHuiJuan |
// * @author: wuHuiJuan
|
||||
* @create: 2019/12/27 15:16 |
// * @create: 2019/12/27 15:16
|
||||
*/ |
// */
|
||||
@Slf4j |
//@Slf4j
|
||||
@Component |
//@Component
|
||||
public class DelayProducer { |
//public class DelayProducer {
|
||||
@Autowired |
// @Autowired
|
||||
private RabbitTemplate rabbitTemplate; |
// private RabbitTemplate rabbitTemplate;
|
||||
|
//
|
||||
//消息发送后的回调函数
|
// //消息发送后的回调函数
|
||||
/** |
// /**
|
||||
* 生产者回调函数：confirm确认消息投递成功 |
// * 生产者回调函数：confirm确认消息投递成功
|
||||
*/ |
// */
|
||||
final RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> { |
// final RabbitTemplate.ConfirmCallback confirmCallback = (correlationData, ack, cause) -> {
|
||||
|
//
|
||||
String messageId = correlationData.getId(); |
// String messageId = correlationData.getId();
|
||||
if (ack) { |
// if (ack) {
|
||||
System.out.println("消息发送成功" + correlationData); |
// System.out.println("消息发送成功" + correlationData);
|
||||
log.info("消息投递成功，{}",messageId); |
// log.info("消息投递成功，{}",messageId);
|
||||
//进行消息记录的数据库更新
|
// //进行消息记录的数据库更新
|
||||
|
//
|
||||
}else{ |
// }else{
|
||||
log.info("消息投递失败"); |
// log.info("消息投递失败");
|
||||
} |
// }
|
||||
|
//
|
||||
}; |
// };
|
||||
|
//
|
||||
/** |
// /**
|
||||
* 通过延迟消息插件发动延迟消息 |
// * 通过延迟消息插件发动延迟消息
|
||||
* @param msg |
// * @param msg
|
||||
* @param expiration |
// * @param expiration
|
||||
*/ |
// */
|
||||
public void sendDelayMessage(Object msg, Long expiration){ |
// public void sendDelayMessage(Object msg, Long expiration){
|
||||
|
//
|
||||
//绑定异步监听回调函数
|
// //绑定异步监听回调函数
|
||||
rabbitTemplate.setConfirmCallback(confirmCallback); |
// rabbitTemplate.setConfirmCallback(confirmCallback);
|
||||
|
//
|
||||
rabbitTemplate.convertAndSend(RabbitMQConfig.GAME_STATUS_DELAY_SEND_EXCHANGE,RabbitMQConfig.GAME_STATUS_DELAY_SEND_ROUTING_KEY, msg,(message)->{ |
// rabbitTemplate.convertAndSend(RabbitMQConfig.GAME_STATUS_DELAY_SEND_EXCHANGE,RabbitMQConfig.GAME_STATUS_DELAY_SEND_ROUTING_KEY, msg,(message)->{
|
||||
message.getMessageProperties().setHeader("x-delay", expiration);//设置延迟时间
|
// message.getMessageProperties().setHeader("x-delay", expiration);//设置延迟时间
|
||||
return message; |
// return message;
|
||||
}, new CorrelationData("123")); |
// }, new CorrelationData("123"));
|
||||
} |
// }
|
||||
} |
//}
|
||||
|
|||||
Loading…
Reference in new issue