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

mq-将送礼和幸运礼物的同步批量发送改成异步发送

Browse Source
This commit is contained in:
khalil
2025-05-13 17:44:52 +08:00
parent f9708a9d3b
commit 6d66be2369
2 changed files with 48 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/mq/RocketMQService.java
View File

@@ -2,14 +2,10 @@ package com.accompany.business.service.mq;
import com.accompany.business.message.*;
import com.accompany.business.message.linearlypool.LinearlyPoolPrizeMessage;
import com.accompany.business.model.miniGame.MiniGameMatchRound;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.producer.MQMessageProducer;
import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
@@ -26,8 +22,6 @@ import java.util.stream.Collectors;
@Service
public class RocketMQService {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Autowired
private MQMessageProducer mqMessageProducer;
@@ -49,12 +43,9 @@ public class RocketMQService {
List<Message<String>> messageList = giftMessages.stream()
.map(giftMessage -> MessageBuilder.withPayload(JSON.toJSONString(giftMessage)).build())
.collect(Collectors.toList());
SendResult sendResult = rocketMQTemplate.syncSend(MqConstant.GIFT_TOPIC, messageList);
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())){
log.info("sendGiftMessage success result: {} message: {}", JSON.toJSONString(sendResult), messageList);
} else {
log.error("sendGiftMessage fail result: {} message: {}", JSON.toJSONString(sendResult), messageList);
}
mqMessageProducer.send(MqConstant.GIFT_TOPIC, messageList,
sendResult -> log.info("sendGiftMessage success result: {} message: {}", JSON.toJSONString(sendResult), messageList),
throwable -> log.error("sendGiftMessage fail message: {}", messageList, throwable));
}
/**
@@ -66,24 +57,18 @@ public class RocketMQService {
List<Message<String>> messageList = lucky24Messages.stream()
.map(giftMessage -> MessageBuilder.withPayload(JSON.toJSONString(giftMessage)).build())
.collect(Collectors.toList());
SendResult sendResult = rocketMQTemplate.syncSend(MqConstant.LUCKY_24_TOPIC, messageList);
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())){
log.info("sendLucky24Message success result: {} message: {}", JSON.toJSONString(sendResult), messageList);
} else {
log.error("sendLucky24Message fail result: {} message: {}", JSON.toJSONString(sendResult), messageList);
}
mqMessageProducer.send(MqConstant.LUCKY_24_TOPIC, messageList,
sendResult -> log.info("sendLucky24Message success result: {} message: {}", JSON.toJSONString(sendResult), messageList),
throwable -> log.error("sendLucky24Message fail message: {}", messageList, throwable));
}
public void sendBatchBravoMessage(List<BravoMessage> bravoMessages) {
List<Message<String>> messageList = bravoMessages.stream()
.map(giftMessage -> MessageBuilder.withPayload(JSON.toJSONString(giftMessage)).build())
.collect(Collectors.toList());
SendResult sendResult = rocketMQTemplate.syncSend(MqConstant.BRAVO_TOPIC, messageList);
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())){
log.info("sendBravoMessage success result: {} message: {}", JSON.toJSONString(sendResult), messageList);
} else {
log.error("sendBravoMessage fail result: {} message: {}", JSON.toJSONString(sendResult), messageList);
}
mqMessageProducer.send(MqConstant.BRAVO_TOPIC, messageList,
sendResult -> log.info("sendBravoMessage success result: {} message: {}", JSON.toJSONString(sendResult), messageList),
throwable -> log.error("sendBravoMessage fail message: {}", messageList, throwable));
}
/**
@@ -95,12 +80,9 @@ public class RocketMQService {
List<Message<String>> messageList = lucky24Messages.stream()
.map(giftMessage -> MessageBuilder.withPayload(JSON.toJSONString(giftMessage)).build())
.collect(Collectors.toList());
SendResult sendResult = rocketMQTemplate.syncSend(MqConstant.LUCKY_25_TOPIC, messageList);
if (SendStatus.SEND_OK.equals(sendResult.getSendStatus())){
log.info("sendLucky25Message success result: {} message: {}", JSON.toJSONString(sendResult), messageList);
} else {
log.error("sendLucky25Message fail result: {} message: {}", JSON.toJSONString(sendResult), messageList);
}
mqMessageProducer.send(MqConstant.LUCKY_25_TOPIC, messageList,
sendResult -> log.info("sendLucky25Message success result: {} message: {}", JSON.toJSONString(sendResult), messageList),
throwable -> log.error("sendLucky25Message fail message: {}", messageList, throwable));
}
/**

36
accompany-mq/accompany-mq-service/src/main/java/com/accompany/mq/producer/MQMessageProducer.java
View File

@@ -6,8 +6,11 @@ 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.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.CollectionUtils;
import java.util.Collection;
import java.util.function.Consumer;
/**
@@ -40,6 +43,39 @@ public class MQMessageProducer {
send(queueName, object, success, error, null, delayLevel);
}
/**
* @param queueName
* @param success
* @param error
* @param <T>
*/
public <T extends Message> void send(String queueName, Collection<T> objectList, Consumer<SendResult> success, Consumer<Throwable> error) {
if (CollectionUtils.isEmpty(objectList)) {
return;
}
long timeout = rocketMQTemplate.getProducer().getSendMsgTimeout();
try {
rocketMQTemplate.asyncSend(queueName, objectList, 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);
} catch (Exception e) {
log.error(e.getMessage(), e);
}
}
/**
* @param queueName
* @param object