moliparty/peko-java
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

rocketmq消费者和生产者代码拆分

Browse Source
This commit is contained in:
liaozetao
2023-12-13 14:19:24 +08:00
parent e9f7d91c1a
commit e5d7ba84cd
44 changed files with 1163 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
accompany-mq/accompany-mq-sdk/pom.xml Normal file
View File

@@ -0,0 +1,21 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.accompany</groupId>
<artifactId>accompany-mq</artifactId>
<version>1.0.0</version>
</parent>
<artifactId>accompany-mq-sdk</artifactId>
<dependencies>
<dependency>
<groupId>com.accompany</groupId>
<artifactId>accompany-core</artifactId>
<version>${revision}</version>
</dependency>
</dependencies>
</project>

41
accompany-mq/accompany-mq-sdk/src/main/java/com/accompany/mq/constant/MqConstant.java Normal file
View File

@@ -0,0 +1,41 @@
package com.accompany.mq.constant;
public interface MqConstant {
String GIFT_TOPIC = "gift_topic";
String GIFT_CONSUME_GROUP = "gift_consume_group";
String OPEN_BOX_TOPIC = "open_box_topic";
String OPEN_BOX_CONSUME_GROUP = "open_box_consume_group";
String LINEARLY_POOL_TOPIC = "linearly_pool_topic";
String LINEARLY_POOL_CONSUME_GROUP = "linearly_pool_consume_group";
String CLEAN_MUSIC_TOPIC = "clean_music_topic";
String CLEAN_MUSIC_CONSUME_GROUP = "clean_music_consume_group";
String ACTIVITY_PACK_TOPIC = "activity_pack_topic";
String ACTIVITY_PACK_CONSUME_GROUP = "activity_pack_consume_group";
String SIGN_DRAW_GOLD_TOPIC = "sign_draw_gold_topic";
String SIGN_DRAW_GOLD_CONSUME_GROUP = "sign_draw_gold_consume_group";
String SIGN_TOPIC = "sign_topic";
String SIGN_CONSUME_GROUP = "sign_consume_group";
String RADISH_GIFT_TOPIC = "radish_gift_topic";
String RADISH_GIFT_CONSUME_GROUP = "radish_gift_consume_group";
String PAY_FINISH_TOPIC = "pay_finish_topic";
String PAY_FINISH_CONSUME_GROUP = "pay_finish_consume_group";
String USER_FIRST_LOGIN_TOPIC = "user_first_login_topic";
String USER_FIRST_LOGIN_CONSUME_GROUP = "user_first_login_consume_group";
String VOICE_LIKE_TOPIC = "voice_like_topic";
String VOICE_LIKE_CONSUME_GROUP = "voice_like_consume_group";
String YI_DUN_TEXT_ANTI_TOPIC = "yidun_text_anti_topic";
String YI_DUN_TEXT_ANTI_CONSUME_GROUP = "yidun_text_anti_consume_group";
}

25
accompany-mq/accompany-mq-sdk/src/main/java/com/accompany/mq/model/BaseMqMessage.java Normal file
View File

@@ -0,0 +1,25 @@
package com.accompany.mq.model;
import lombok.Data;
import java.io.Serializable;
/**
* @author: liaozetao
* @date: 2023/12/1 14:28
* @description:
*/
@Data
public class BaseMqMessage implements Serializable {
// /**
// * 消息唯一标识
// */
// private String messId;
//
// /**
// * 消息创建时间
// */
// private Long messTime;
}

32
accompany-mq/accompany-mq-service/pom.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.accompany</groupId>
<artifactId>accompany-mq</artifactId>
<version>1.0.0</version>
</parent>
<artifactId>accompany-mq-service</artifactId>
<dependencies>
<dependency>
<groupId>com.accompany</groupId>
<artifactId>accompany-core</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>com.accompany</groupId>
<artifactId>accompany-mq-sdk</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot</artifactId>
<version>${rocketmq-spring-boot.version}</version>
</dependency>
</dependencies>
</project>

60
accompany-mq/accompany-mq-service/src/main/java/com/accompany/mq/listener/AbstractMessageListener.java Normal file
View File

@@ -0,0 +1,60 @@
package com.accompany.mq.listener;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.model.BaseMqMessage;
import com.google.gson.internal.$Gson$Types;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.factory.annotation.Autowired;
import java.lang.reflect.ParameterizedType;
import java.util.UUID;
/**
* @author: liaozetao
* @date: 2023/12/13 10:38
* @description:
*/
@Slf4j
public abstract class AbstractMessageListener<T extends BaseMqMessage> implements RocketMQListener<String> {
private static final int MQ_LOCK_SECONDS = 30 * 60;
@Autowired
protected JedisService jedisService;
@Override
public void onMessage(String message) {
try {
log.info("====mq message start====");
log.info("text message : {}", message);
if (!message.startsWith(StrUtil.DELIM_START) || !message.endsWith(StrUtil.DELIM_END)) {
return;
}
T mqMessage = JSONUtil.toBean(message, $Gson$Types.canonicalize(((ParameterizedType) this.getClass().getGenericSuperclass()).getActualTypeArguments()[0]), true);
if (mqMessage == null) {
return;
}
//String messId = mqMessage.getMessId();
String messId = UUID.randomUUID().toString();
//防止消息被重复消费
RedisKey mqLock = mqLock();
if (mqLock != null && !jedisService.setnx(mqLock.getKey(messId), Boolean.TRUE.toString(), MQ_LOCK_SECONDS)) {
log.error("mq lock : {}, message had handle, msg : {}", mqLock.getKey(messId), message);
return;
}
onMessage(mqMessage);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
protected abstract void onMessage(T object) throws Exception;
protected RedisKey mqLock() {
return null;
}
}

158
accompany-mq/accompany-mq-service/src/main/java/com/accompany/mq/producer/MQMessageProducer.java Normal file
View File

@@ -0,0 +1,158 @@
package com.accompany.mq.producer;
import cn.hutool.core.lang.UUID;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.accompany.mq.model.BaseMqMessage;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import java.util.function.Consumer;
/**
* @author: liaozetao
* @date: 2023/12/13 10:39
* @description:
*/
@Slf4j
public class MQMessageProducer {
private final RocketMQTemplate rocketMQTemplate;
public MQMessageProducer(RocketMQTemplate rocketMQTemplate) {
this.rocketMQTemplate = rocketMQTemplate;
}
public <T extends BaseMqMessage> void send(String queueName, T object) {
send(queueName, object, null, null, null, null);
}
public <T extends BaseMqMessage> void send(String queueName, T object, Integer delayLevel) {
send(queueName, object, null, null, null, delayLevel);
}
public <T extends BaseMqMessage> void send(String queueName, T object, Consumer<SendResult> success, Consumer<Throwable> error) {
send(queueName, object, success, error, null, null);
}
public <T extends BaseMqMessage> void send(String queueName, T object, Consumer<SendResult> success, Consumer<Throwable> error, Integer delayLevel) {
send(queueName, object, success, error, null, delayLevel);
}
/**
* @param queueName
* @param object
* @param success
* @param error
* @param timeout
* @param delayLevel 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 对应定时的延时时间
* @param <T>
*/
public <T extends BaseMqMessage> void send(String queueName, T object, Consumer<SendResult> success, Consumer<Throwable> error, Integer timeout, Integer delayLevel) {
if (object == null) {
return;
}
if (timeout == null) {
timeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
}
if (delayLevel == null) {
delayLevel = 0;
}
try {
// String messId = object.getMessId();
// if (StrUtil.isEmpty(messId)) {
// messId = UUID.randomUUID().toString();
// }
// object.setMessId(messId);
// object.setMessTime(System.currentTimeMillis());
String objectJson = JSONUtil.toJsonStr(object);
log.info("queueName : {}, message : {}", queueName, objectJson);
rocketMQTemplate.asyncSend(queueName, MessageBuilder.withPayload(objectJson).build(), new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
if (success != null) {
success.accept(sendResult);
}
}
@Override
public void onException(Throwable throwable) {
if (error != null) {
error.accept(throwable);
}
}
}, timeout, delayLevel);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
public <T extends BaseMqMessage> void sendOrderly(String queueName, T object, String hashKey) {
sendOrderly(queueName, object, hashKey, null, null, null, null);
}
public <T extends BaseMqMessage> void sendOrderly(String queueName, T object, String hashKey, Integer delayLevel) {
sendOrderly(queueName, object, hashKey, null, null, null, delayLevel);
}
public <T extends BaseMqMessage> void sendOrderly(String queueName, T object, String hashKey, Consumer<SendResult> success, Consumer<Throwable> error) {
sendOrderly(queueName, object, hashKey, success, error, null, null);
}
public <T extends BaseMqMessage> void sendOrderly(String queueName, T object, String hashKey, Consumer<SendResult> success, Consumer<Throwable> error, Integer delayLevel) {
sendOrderly(queueName, object, hashKey, success, error, null, delayLevel);
}
/**
* @param queueName
* @param object
* @param hashKey
* @param success
* @param error
* @param timeout
* @param delayLevel 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 对应定时的延时时间
* @param <T>
*/
public <T extends BaseMqMessage> void sendOrderly(String queueName, T object, String hashKey, Consumer<SendResult> success, Consumer<Throwable> error, Integer timeout, Integer delayLevel) {
if (object == null) {
return;
}
if (timeout == null) {
timeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
}
if (delayLevel == null) {
delayLevel = 0;
}
try {
// String messId = object.getMessId();
// if (StrUtil.isEmpty(messId)) {
// messId = UUID.randomUUID().toString();
// }
// object.setMessId(messId);
// object.setMessTime(System.currentTimeMillis());
String objectJson = JSONUtil.toJsonStr(object);
log.info("queueName : {}, message : {}", queueName, objectJson);
rocketMQTemplate.asyncSendOrderly(queueName, MessageBuilder.withPayload(objectJson).build(), hashKey, new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
if (success != null) {
success.accept(sendResult);
}
}
@Override
public void onException(Throwable throwable) {
if (error != null) {
error.accept(throwable);
}
}
}, timeout, delayLevel);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
}

32
accompany-mq/accompany-mq-web/pom.xml Normal file
View File

@@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.accompany</groupId>
<artifactId>accompany-mq</artifactId>
<version>1.0.0</version>
</parent>
<artifactId>accompany-mq-web</artifactId>
<dependencies>
<dependency>
<groupId>com.accompany</groupId>
<artifactId>accompany-core-starter</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>com.accompany</groupId>
<artifactId>accompany-mq-service</artifactId>
<version>${revision}</version>
</dependency>
<dependency>
<groupId>com.accompany</groupId>
<artifactId>accompany-business-service</artifactId>
<version>${revision}</version>
</dependency>
</dependencies>
</project>

24
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/configuration/RocketMQConfiguration.java Normal file
View File

@@ -0,0 +1,24 @@
package com.accompany.mq.configuration;
import com.accompany.mq.producer.MQMessageProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
/**
* @author: liaozetao
* @date: 2023/12/13 11:10
* @description:
*/
@Slf4j
@Configuration
@ConditionalOnClass(value = RocketMQTemplate.class)
public class RocketMQConfiguration {
@Bean
public MQMessageProducer mqMessageProducer(RocketMQTemplate rocketMQTemplate) {
return new MQMessageProducer(rocketMQTemplate);
}
}

44
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/ActivityPackMessageConsumer.java Normal file
View File

@@ -0,0 +1,44 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.ActivityPackMessage;
import com.accompany.business.service.activity.ActivityPackMessageService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.utils.BlankUtil;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 礼包消息监听器
*
* @author xiaoyuyou
* @date 2018/9/5 11:03
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.ACTIVITY_PACK_TOPIC, consumerGroup = MqConstant.ACTIVITY_PACK_CONSUME_GROUP)
public class ActivityPackMessageConsumer extends AbstractMessageListener<ActivityPackMessage> {
@Autowired
private JedisService jedisService;
@Autowired
private ActivityPackMessageService activityPackMessageService;
@Override
public void onMessage(ActivityPackMessage packMessage) {
log.info("handle activity pack message {}", JSON.toJSONString(packMessage));
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.mq_pack_status.getKey(), packMessage.getMessId());
if (BlankUtil.isBlank(messStatus)) {
return;
}
activityPackMessageService.handleActivityPackMessage(packMessage);
}
}

55
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/CleanMusicDelayMessageConsumer.java Normal file
View File

@@ -0,0 +1,55 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.CleanMusicDelayMessage;
import com.accompany.business.service.ktv.UserChooseMusicService;
import com.accompany.business.service.user.UserInRoomService;
import com.accompany.business.vo.RoomVo;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 清歌消息监听器
* @author xiaoyuyou
* @date 2018/9/5 11:03
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.CLEAN_MUSIC_TOPIC, consumerGroup = MqConstant.CLEAN_MUSIC_CONSUME_GROUP)
public class CleanMusicDelayMessageConsumer extends AbstractMessageListener<CleanMusicDelayMessage> {
@Autowired
private UserChooseMusicService userChooseMusicService;
@Autowired
private UserInRoomService userInRoomService;
@Autowired
private JedisService jedisService;
@SneakyThrows
@Override
public void onMessage(CleanMusicDelayMessage cleanMusicMsg) {
log.info("clean music delay message {}", JSON.toJSONString(cleanMusicMsg));
Long roomUid = cleanMusicMsg.getRoomUid();
Long uid = cleanMusicMsg.getUid();
String messId = jedisService.hget(RedisKey.room_clean_music_messId.getKey(), roomUid + "_" + uid);
log.info("clean music listener exec, roomUid:{}, uid:{}, period:{}, redis messId:{}, current messId:{}",
roomUid, uid, System.currentTimeMillis() - cleanMusicMsg.getMessTime(), messId, cleanMusicMsg.getMessId());
if (StringUtils.isEmpty(messId) || !StringUtils.equals(messId, cleanMusicMsg.getMessId())){
return;
}
RoomVo roomVo = userInRoomService.getUserInRoomInfoCache(uid);
if (roomVo == null || roomVo.getUid().longValue() != roomUid.longValue()){
userChooseMusicService.deleteUserAllChooseMusic(roomUid, uid);
}
}
}

38
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/GiftMessageConsumer.java Normal file
View File

@@ -0,0 +1,38 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.GiftMessage;
import com.accompany.business.service.gift.GiftMessageService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.utils.BlankUtil;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.GIFT_TOPIC, consumerGroup = MqConstant.GIFT_CONSUME_GROUP)
public class GiftMessageConsumer extends AbstractMessageListener<GiftMessage> {
@Autowired
private JedisService jedisService;
@Autowired
private GiftMessageService giftMessageService;
@Override
public void onMessage(GiftMessage giftMessage) {
log.info("onMessage giftMessage: {}", giftMessage.toString());
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.mq_gift_status.getKey(), giftMessage.getMessId());
if (BlankUtil.isBlank(messStatus)) {
return;
}
giftMessageService.handleGiftMessage(giftMessage);
}
}

41
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/LinearlyPoolPrizeMessageConsumer.java Normal file
View File

@@ -0,0 +1,41 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.linearlypool.LinearlyPoolPrizeMessage;
import com.accompany.business.service.linearlypool.LinearlyPoolPrizeMessageService;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 线性奖池中奖消息
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.LINEARLY_POOL_TOPIC, consumerGroup = MqConstant.LINEARLY_POOL_CONSUME_GROUP, consumeMode = ConsumeMode.ORDERLY)
public class LinearlyPoolPrizeMessageConsumer extends AbstractMessageListener<LinearlyPoolPrizeMessage> {
@Autowired
private LinearlyPoolPrizeMessageService prizeMessageService;
@Autowired
private JedisService jedisService;
@Override
public void onMessage(LinearlyPoolPrizeMessage message) {
Thread thread = Thread.currentThread();
log.info("{} thread handle linearly-pool-draw-queue message {}", thread.getName(), JSON.toJSONString(message));
prizeMessageService.onMessage(message);
// 删除消息标识
jedisService.hdel(RedisKey.mq_linearly_pool_prize_status.getKey(), message.getMessId());
}
}

41
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/OpenBoxPrizeMessageConsumer.java Normal file
View File

@@ -0,0 +1,41 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.BoxPrizeMessage;
import com.accompany.business.service.box.BoxPrizeMessageService;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* Created by PaperCut on 2018/7/16.
* 开箱子中奖消息
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.OPEN_BOX_TOPIC, consumerGroup = MqConstant.OPEN_BOX_CONSUME_GROUP, consumeMode = ConsumeMode.ORDERLY)
public class OpenBoxPrizeMessageConsumer extends AbstractMessageListener<BoxPrizeMessage> {
@Autowired
private BoxPrizeMessageService boxPrizeMessageService;
@Autowired
private JedisService jedisService;
@Override
public void onMessage(BoxPrizeMessage message) {
Thread thread = Thread.currentThread();
log.info("{} thread handle openbox-queue message {}", thread.getName(), JSON.toJSONString(message));
boxPrizeMessageService.onMessage(message);
// 删除消息标识
jedisService.hdel(RedisKey.mq_prize_status.getKey(), message.getMessId());
}
}

56
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/PayFinishMessageConsumer.java Normal file
View File

@@ -0,0 +1,56 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.PayFinishMessage;
import com.accompany.business.service.activity.ChargeActivityPackRecordService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.utils.BlankUtil;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.accompany.payment.model.ChargeRecord;
import com.alibaba.fastjson.JSON;
import com.google.gson.Gson;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 处理支付完成后的一些操作
* 如：
* 1. 充值送礼包：充值完成后，判断用户的充值金额、次数等，并发放相应的礼包
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.PAY_FINISH_TOPIC, consumerGroup = MqConstant.PAY_FINISH_CONSUME_GROUP)
public class PayFinishMessageConsumer extends AbstractMessageListener<PayFinishMessage> {
@Autowired
private JedisService jedisService;
@Autowired
private ChargeActivityPackRecordService chargeActivityPackRecordService;
private final Gson gson = new Gson();
@Override
public void onMessage(PayFinishMessage payFinishMessage) {
log.info("onMessage payFinishMessage: {}", JSON.toJSONString(payFinishMessage));
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.mq_pay_finish_status.getKey(), payFinishMessage.getMessId());
if (BlankUtil.isBlank(messStatus)) {
log.info("消息已处理，消息id:{}", payFinishMessage.getMessId());
return;
}
//处理具体逻辑
ChargeRecord chargeRecord = gson.fromJson(messStatus, ChargeRecord.class);
//处理首充礼包
chargeActivityPackRecordService.handleFirstChargeActivityPack(chargeRecord);
//消费完成，删除该消息
jedisService.hdel(RedisKey.mq_pay_finish_status.getKey(), payFinishMessage.getMessId());
}
}

38
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/RadishGiftMessageConsumer.java Normal file
View File

@@ -0,0 +1,38 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.RadishGiftMessage;
import com.accompany.business.service.gift.RadishGiftMessageService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.utils.BlankUtil;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.RADISH_GIFT_TOPIC, consumerGroup = MqConstant.RADISH_GIFT_CONSUME_GROUP)
public class RadishGiftMessageConsumer extends AbstractMessageListener<RadishGiftMessage> {
@Autowired
private JedisService jedisService;
@Autowired
private RadishGiftMessageService radishGiftMessageService;
@Override
public void onMessage(RadishGiftMessage radishGiftMessage) {
log.info("onMessage radishGiftMessage: {}", radishGiftMessage.toString());
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.mq_radish_gift_status.getKey(), radishGiftMessage.getMessId());
if (BlankUtil.isBlank(messStatus)) {
return;
}
radishGiftMessageService.handleRadishGiftMessage(radishGiftMessage);
}
}

45
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/SignDrawGoldMessageConsumer.java Normal file
View File

@@ -0,0 +1,45 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.SignDrawGoldMessage;
import com.accompany.business.service.signweb.SignDrawGoldService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.utils.BlankUtil;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 瓜分金币
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.SIGN_DRAW_GOLD_TOPIC, consumerGroup = MqConstant.SIGN_DRAW_GOLD_CONSUME_GROUP)
public class SignDrawGoldMessageConsumer extends AbstractMessageListener<SignDrawGoldMessage> {
@Autowired
private JedisService jedisService;
@Autowired
private SignDrawGoldService signDrawGoldService;
@SneakyThrows
@Override
public void onMessage(SignDrawGoldMessage signDrawGoldMessage) {
log.info("onMessage signDrawGoldMessage: {}", JSON.toJSONString(signDrawGoldMessage));
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.mq_sign_draw_gold_status.getKey(), signDrawGoldMessage.getMessId());
if (BlankUtil.isBlank(messStatus)) {
return;
}
signDrawGoldService.handleDrawGoldMessage(signDrawGoldMessage);
}
}

43
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/SignMessageConsumer.java Normal file
View File

@@ -0,0 +1,43 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.SignMessage;
import com.accompany.business.service.signweb.SignService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.utils.BlankUtil;
import com.accompany.core.service.common.JedisService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 每日签到奖励发放
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.SIGN_TOPIC, consumerGroup = MqConstant.SIGN_CONSUME_GROUP)
public class SignMessageConsumer extends AbstractMessageListener<SignMessage> {
@Autowired
private JedisService jedisService;
@Autowired
private SignService signService;
@Override
public void onMessage(SignMessage signMessage) {
log.info("onMessage signMessage: {}", JSON.toJSONString(signMessage));
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.mq_sign_status.getKey(), signMessage.getMessId());
if (BlankUtil.isBlank(messStatus)) {
return;
}
signService.handlerSign(signMessage);
}
}

79
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/UserFirstLoginMessageConsumer.java Normal file
View File

@@ -0,0 +1,79 @@
package com.accompany.mq.consumer;
import com.accompany.business.service.gamemange.GameManageAccessTicketBizService;
import com.accompany.common.redis.RedisKey;
import com.accompany.common.utils.BlankUtil;
import com.accompany.core.exception.ServiceException;
import com.accompany.core.service.common.JedisService;
import com.accompany.core.vo.UserFirstLoginMsgVO;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import com.xuanyin.gamematch.constants.GameManageAccessTicketEnum;
import com.xuanyin.gamematch.handler.ticketaccessstatey.ITicketAccessStategy;
import com.xuanyin.gamematch.handler.ticketaccessstatey.TicketAccessStategyFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.USER_FIRST_LOGIN_TOPIC, consumerGroup = MqConstant.USER_FIRST_LOGIN_CONSUME_GROUP)
public class UserFirstLoginMessageConsumer extends AbstractMessageListener<UserFirstLoginMsgVO> {
@Autowired
private JedisService jedisService;
@Autowired
private TicketAccessStategyFactory factory;
@Autowired
private GameManageAccessTicketBizService gameManageAccessTicketBizService;
@Override
public void onMessage(UserFirstLoginMsgVO msg) {
log.info("用户首登消息队列, msg: {} ", JSON.toJSONString(msg));
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.mq_user_first_login_status.getKey(), msg.getUid().toString());
if (BlankUtil.isBlank(messStatus)) {
return;
}
handleDayFirstLogin(msg);
handleWeekFirstLogin(msg);
jedisService.hdel(RedisKey.mq_user_first_login_status.getKey(), msg.getUid().toString());
}
@Async
void handleDayFirstLogin(UserFirstLoginMsgVO msg) {
if (msg.getIsDayFirstLogin()) {
try {
log.info("处理用户每日首登 uid {}", msg.getUid());
ITicketAccessStategy stategy = factory.getInstance(GameManageAccessTicketEnum.DAY_LOGIN.getValue());
stategy.doSend(msg.getUid());
gameManageAccessTicketBizService.setTicketAccessInfoToCache(stategy.getAccessType().getValue(), msg.getUid());
} catch (ServiceException ignored){
}
}
}
@Async
void handleWeekFirstLogin(UserFirstLoginMsgVO msg) {
if (msg.getIsWeekFirstLogin()) {
try {
log.info("处理用户每周首登 uid {}", msg.getUid());
ITicketAccessStategy stategy = factory.getInstance(GameManageAccessTicketEnum.WEEK_LOGIN.getValue());
stategy.doSend(msg.getUid());
gameManageAccessTicketBizService.setTicketAccessInfoToCache(stategy.getAccessType().getValue(), msg.getUid());
} catch (ServiceException ignored){
}
}
}
}

46
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/VoiceLikeMessageConsumer.java Normal file
View File

@@ -0,0 +1,46 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.VoiceLikeMessage;
import com.accompany.business.service.voice.VoiceService;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisService;
import com.accompany.core.util.StringUtils;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* @author chucheng
* @date 2019-06-05
* @description 声音瓶子 喜欢/不喜欢消息监听
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.VOICE_LIKE_TOPIC, consumerGroup = MqConstant.VOICE_LIKE_CONSUME_GROUP)
public class VoiceLikeMessageConsumer extends AbstractMessageListener<VoiceLikeMessage> {
@Autowired
private VoiceService voiceService;
@Autowired
private JedisService jedisService;
@Override
public void onMessage(VoiceLikeMessage message) {
log.info("onMessage VoiceLikeMessage: {}", JSON.toJSONString(message));
// 判断该消息是否已经消费过
String messStatus = jedisService.hget(RedisKey.voice_like_status.getKey(), message.getMessId());
if (StringUtils.isBlank(messStatus)) {
log.error("handleVoiceLikeMessage status error.message = {}", JSON.toJSONString(message));
return;
}
voiceService.handleVoiceLikeMessage(message);
}
}

43
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/YiDunIMTextAntiMessageConsumer.java Normal file
View File

@@ -0,0 +1,43 @@
package com.accompany.mq.consumer;
import com.accompany.business.message.YidunIMAntiMessage;
import com.accompany.business.service.netease.YidunAntiHandleService;
import com.accompany.common.redis.RedisKey;
import com.accompany.core.service.common.JedisService;
import com.accompany.core.util.StringUtils;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
/**
* 易盾文本反垃圾消息处理
*/
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.YI_DUN_TEXT_ANTI_TOPIC, consumerGroup = MqConstant.YI_DUN_TEXT_ANTI_CONSUME_GROUP)
public class YiDunIMTextAntiMessageConsumer extends AbstractMessageListener<YidunIMAntiMessage> {
@Autowired
private YidunAntiHandleService yidunAntiHandleService;
@Autowired
private JedisService jedisService;
@Override
public void onMessage(YidunIMAntiMessage message) {
log.info("handle yidun im text anti message {}", JSON.toJSONString(message));
String messStatus = jedisService.hget(RedisKey.voice_like_status.getKey(), message.getMessId());
if (StringUtils.isBlank(messStatus)) {
log.error("handleYidunIMAntiMessage status error.message = {}", JSON.toJSONString(message));
return;
}
yidunAntiHandleService.handlIMTextAnti(message.getChatMsg());
// 删除消息标识
jedisService.hdel(RedisKey.mq_yindun_im_text_anti_status.getKey(), message.getMessId());
}
}

27
accompany-mq/pom.xml Normal file
View File

@@ -0,0 +1,27 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.accompany</groupId>
<artifactId>accompany-dependencies</artifactId>
<version>1.0.0</version>
<relativePath>../accompany-dependencies</relativePath>
</parent>
<artifactId>accompany-mq</artifactId>
<packaging>pom</packaging>
<modules>
<module>accompany-mq-sdk</module>
<module>accompany-mq-service</module>
<module>accompany-mq-web</module>
</modules>
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
</project>