forked from rongchao/epmet-cloud-rizhao
jianjun
4 years ago
10 changed files with 221 additions and 47 deletions
@ -1,23 +0,0 @@ |
|||||
package com.epmet.mq; |
|
||||
|
|
||||
import lombok.extern.slf4j.Slf4j; |
|
||||
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; |
|
||||
import org.apache.rocketmq.spring.core.RocketMQListener; |
|
||||
|
|
||||
/** |
|
||||
* desc: |
|
||||
* |
|
||||
* @author: LiuJanJun |
|
||||
* @date: 2021/4/22 10:10 下午 |
|
||||
* @version: 1.0 |
|
||||
*/ |
|
||||
@Slf4j |
|
||||
//@RocketMQMessageListener(topic = "${game.server.config.business-game-message-topic}", consumerGroup = "gateway-message-consumer-group")
|
|
||||
@RocketMQMessageListener(topic = "group_achievement", consumerGroup = "group") |
|
||||
public class AbstractRocketConsumer implements RocketMQListener<String> { |
|
||||
|
|
||||
@Override |
|
||||
public void onMessage(String s) { |
|
||||
log.info("receive msg:{}",s); |
|
||||
} |
|
||||
} |
|
||||
@ -0,0 +1,37 @@ |
|||||
|
package com.epmet.mq; |
||||
|
|
||||
|
import com.alibaba.fastjson.JSON; |
||||
|
import com.epmet.commons.rocketmq.messages.GroupAchievementMQMsg; |
||||
|
import com.epmet.modules.group.service.StatsAchievementService; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.commons.lang3.StringUtils; |
||||
|
import org.apache.rocketmq.spring.core.RocketMQListener; |
||||
|
import org.springframework.beans.factory.annotation.Autowired; |
||||
|
|
||||
|
/** |
||||
|
* desc: 小组成就mq监听 |
||||
|
* |
||||
|
* @author: LiuJanJun |
||||
|
* @date: 2021/4/22 10:10 下午 |
||||
|
* @version: 1.0 |
||||
|
*/ |
||||
|
@Slf4j |
||||
|
//@Component
|
||||
|
//@RocketMQMessageListener(topic = "group_achievement", consumerGroup = ConsomerGroupConstants.GROUP_ACHIEVEMENT_COMPONENTS_GROUP)
|
||||
|
//@Component
|
||||
|
public class GroupAchieventConsumerListener implements RocketMQListener<GroupAchievementMQMsg> { |
||||
|
@Autowired |
||||
|
private StatsAchievementService statsAchievementService; |
||||
|
|
||||
|
@Override |
||||
|
public void onMessage(GroupAchievementMQMsg msg) { |
||||
|
log.info("receive msg:{}", JSON.toJSONString(msg)); |
||||
|
if (StringUtils.isBlank(msg.getAchievementType()) || StringUtils.isBlank(msg.getCustomerId()) |
||||
|
|| StringUtils.isBlank(msg.getGroupId())) { |
||||
|
log.error("consumer fail,msg:{}", msg); |
||||
|
return; |
||||
|
} |
||||
|
Boolean aBoolean = statsAchievementService.calculateAcm(msg.getCustomerId(), msg.getGroupId(), msg.getAchievementType()); |
||||
|
log.info("consumer msg success,{}", aBoolean); |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,77 @@ |
|||||
|
package com.epmet.mq.type1; |
||||
|
|
||||
|
import com.alibaba.fastjson.JSON; |
||||
|
import com.epmet.commons.rocketmq.messages.GroupAchievementMQMsg; |
||||
|
import com.epmet.commons.tools.distributedlock.DistributedLock; |
||||
|
import com.epmet.commons.tools.exception.ExceptionUtils; |
||||
|
import com.epmet.commons.tools.exception.RenException; |
||||
|
import com.epmet.commons.tools.utils.SpringContextUtils; |
||||
|
import com.epmet.modules.group.service.StatsAchievementService; |
||||
|
import org.apache.commons.lang3.StringUtils; |
||||
|
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; |
||||
|
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; |
||||
|
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; |
||||
|
import org.apache.rocketmq.common.message.MessageExt; |
||||
|
import org.redisson.api.RLock; |
||||
|
import org.slf4j.Logger; |
||||
|
import org.slf4j.LoggerFactory; |
||||
|
|
||||
|
import java.util.List; |
||||
|
import java.util.concurrent.TimeUnit; |
||||
|
|
||||
|
/** |
||||
|
* @Description 小组成就-监听器 |
||||
|
* @return |
||||
|
* @author wxz |
||||
|
* @date 2021.03.03 16:10 |
||||
|
*/ |
||||
|
public class GroupAchievementCustomListener implements MessageListenerConcurrently { |
||||
|
|
||||
|
private Logger logger = LoggerFactory.getLogger(getClass()); |
||||
|
|
||||
|
@Override |
||||
|
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) { |
||||
|
try { |
||||
|
msgs.forEach(this::consumeMessage); |
||||
|
} catch (Exception e) { |
||||
|
logger.error("consumeMessage fail,msg:{}",e.getMessage()); |
||||
|
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
||||
|
} |
||||
|
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; |
||||
|
} |
||||
|
|
||||
|
private void consumeMessage(MessageExt messageExt) { |
||||
|
String msg = new String(messageExt.getBody()); |
||||
|
logger.info("receive msg:{}", JSON.toJSONString(msg)); |
||||
|
GroupAchievementMQMsg msgObj = JSON.parseObject(msg, GroupAchievementMQMsg.class); |
||||
|
|
||||
|
DistributedLock distributedLock = null; |
||||
|
RLock lock = null; |
||||
|
try { |
||||
|
distributedLock = SpringContextUtils.getBean(DistributedLock.class); |
||||
|
lock = distributedLock.getLock(String.format("lock:init_customer_org:%s", msgObj.getCustomerId()), |
||||
|
30l, 30l, TimeUnit.SECONDS); |
||||
|
|
||||
|
|
||||
|
|
||||
|
if (StringUtils.isBlank(msgObj.getAchievementType()) || StringUtils.isBlank(msgObj.getCustomerId()) |
||||
|
|| StringUtils.isBlank(msgObj.getGroupId())){ |
||||
|
logger.error("consumer fail,msg:{}",msgObj); |
||||
|
return; |
||||
|
} |
||||
|
Boolean aBoolean = SpringContextUtils.getBean(StatsAchievementService.class).calculateAcm(msgObj.getCustomerId(), msgObj.getGroupId(), msgObj.getAchievementType()); |
||||
|
logger.info("consumer msg success,{}",aBoolean); |
||||
|
} catch (RenException e) { |
||||
|
// 如果是我们手动抛出的异常，说明在业务可控范围内。目前不需要MQ重试
|
||||
|
logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
||||
|
} catch (Exception e) { |
||||
|
// 不是我们自己抛出的异常，可以让MQ重试
|
||||
|
logger.error("【RocketMQ】初始化客户组织失败:".concat(ExceptionUtils.getErrorStackTrace(e))); |
||||
|
throw e; |
||||
|
} finally { |
||||
|
if (distributedLock != null){ |
||||
|
distributedLock.unLock(lock); |
||||
|
} |
||||
|
} |
||||
|
} |
||||
|
} |
||||
@ -0,0 +1,63 @@ |
|||||
|
package com.epmet.mq.type1; |
||||
|
|
||||
|
import com.epmet.commons.rocketmq.constants.ConsomerGroupConstants; |
||||
|
import com.epmet.commons.rocketmq.constants.TopicConstants; |
||||
|
import lombok.extern.slf4j.Slf4j; |
||||
|
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; |
||||
|
import org.apache.rocketmq.client.consumer.listener.MessageListener; |
||||
|
import org.apache.rocketmq.client.exception.MQClientException; |
||||
|
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel; |
||||
|
import org.springframework.beans.factory.annotation.Value; |
||||
|
import org.springframework.stereotype.Component; |
||||
|
|
||||
|
import javax.annotation.PostConstruct; |
||||
|
|
||||
|
@Slf4j |
||||
|
@Component |
||||
|
public class RocketMQConsumerRegister { |
||||
|
|
||||
|
@Value("${rocketmq.name-server}") |
||||
|
private String nameServer; |
||||
|
|
||||
|
/** |
||||
|
* @return |
||||
|
* @Description 注册监听器 |
||||
|
* @author wxz |
||||
|
* @date 2021.03.03 16:09 |
||||
|
*/ |
||||
|
@PostConstruct |
||||
|
public void registerAllListeners() { |
||||
|
try { |
||||
|
register(ConsomerGroupConstants.GROUP_ACHIEVEMENT_COMPONENTS_GROUP, MessageModel.CLUSTERING, TopicConstants.GROUP_ACHIEVEMENT, "*", new GroupAchievementCustomListener()); |
||||
|
} catch (MQClientException e) { |
||||
|
log.error("registerAllListeners exception", e); |
||||
|
} |
||||
|
} |
||||
|
|
||||
|
public void register(String group, MessageModel messageModel, String topic, String subException, MessageListener listener) throws MQClientException { |
||||
|
// 实例化消费者
|
||||
|
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(group); |
||||
|
|
||||
|
// 设置NameServer的地址
|
||||
|
consumer.setNamesrvAddr(nameServer); |
||||
|
consumer.setMessageModel(messageModel); |
||||
|
consumer.setInstanceName(buildInstanceName()); |
||||
|
// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
|
||||
|
consumer.subscribe(topic, subException); |
||||
|
// 注册回调实现类来处理从broker拉取回来的消息
|
||||
|
consumer.registerMessageListener(listener); |
||||
|
// 启动消费者实例
|
||||
|
consumer.start(); |
||||
|
} |
||||
|
|
||||
|
private String buildInstanceName() { |
||||
|
String instanceName = ""; |
||||
|
for (int i = 0; i < 4; i++) { |
||||
|
int t = (int) (Math.random() * 10); |
||||
|
instanceName = instanceName.concat(t + ""); |
||||
|
} |
||||
|
|
||||
|
return instanceName; |
||||
|
} |
||||
|
|
||||
|
} |
||||
Loading…
Reference in new issue