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

账单-mq-尝试通过修改createTime毫秒位，追求批量送礼时createTime顺序不追求主键顺序，修改gift和bill为随机顺序mq

Browse Source
This commit is contained in:
khalil
2025-01-13 22:06:37 +08:00
parent d82f433306
commit 19809f362b
4 changed files with 20 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
accompany-business/accompany-business-service/src/main/java/com/accompany/business/service/gift/GiftSendService.java
View File

@@ -532,6 +532,10 @@ public class GiftSendService extends BaseService {
Long roomUid = null != room? room.getUid(): null;
Byte roomType = null != room? room.getType(): null;
//因为giftMessage是随机顺序
//处理时间毫秒位，方便查询时排序
sendGiftTime = DateTimeUtil.addMilliSecond(sendGiftTime, recvUids.length);
List<GiftMessage> giftMessages = new ArrayList<>(recvUids.length);
for (Long recvUid : recvUids) {
UserPurse afterPurse = null;
@@ -543,6 +547,8 @@ public class GiftSendService extends BaseService {
after.setDiamonds(DoubleUtil.add(after.getDiamonds(), everyGoldNum.doubleValue()));
}
sendGiftTime = DateTimeUtil.addMilliSecond(sendGiftTime, -1);
String id = idGenerator.nextUUID(null);
GiftMessage giftMessage = buildGiftMessage(sender.getUid(), recvUid,
roomUid, gift.getGiftId(), gift.getConsumeType(), gift.getGiftType(),

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

@@ -34,7 +34,13 @@ public class RocketMQService {
* 送消息，发送到MQ
*/
public void sendBillRecordMessage(BillMessage billMessage) {
mqMessageProducer.sendOrderly(MqConstant.BILL_RECORD_TOPIC, billMessage, billMessage.getUid().toString(),
/*mqMessageProducer.sendOrderly(MqConstant.BILL_RECORD_TOPIC, billMessage, billMessage.getUid().toString(),
sendResult -> log.info("sendBillRecordMessage success message: {} queue {}", JSON.toJSONString(billMessage), sendResult.getMessageQueue().getQueueId()),
throwable -> {
log.error("sendBillRecordMessage fail message: {}", JSON.toJSONString(billMessage), throwable);
//SpringContextHolder.getBean(OpenBoxPrizeMessageMQListener.class).onMessage(message);
});*/
mqMessageProducer.send(MqConstant.BILL_RECORD_TOPIC, billMessage,
sendResult -> log.info("sendBillRecordMessage success message: {} queue {}", JSON.toJSONString(billMessage), sendResult.getMessageQueue().getQueueId()),
throwable -> {
log.error("sendBillRecordMessage fail message: {}", JSON.toJSONString(billMessage), throwable);
@@ -48,12 +54,13 @@ public class RocketMQService {
* @param giftMessages
*/
public void sendBatchGiftMessage(List<GiftMessage> giftMessages) {
Long uid = giftMessages.get(0).getRecvUid();
//Long uid = giftMessages.get(0).getRecvUid();
List<Message<String>> messageList = giftMessages.stream()
.map(giftMessage -> MessageBuilder.withPayload(JSON.toJSONString(giftMessage)).build())
.collect(Collectors.toList());
SendResult sendResult = rocketMQTemplate.syncSendOrderly(MqConstant.GIFT_TOPIC, messageList, uid.toString());
//SendResult sendResult = rocketMQTemplate.syncSendOrderly(MqConstant.GIFT_TOPIC, messageList, uid.toString());
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 {

3
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/BillMessageConsumer.java
View File

@@ -14,7 +14,8 @@ import org.springframework.stereotype.Component;
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
@RocketMQMessageListener(topic = MqConstant.BILL_RECORD_TOPIC, consumerGroup = MqConstant.BILL_RECORD_CONSUME_GROUP, consumeMode = ConsumeMode.ORDERLY)
@RocketMQMessageListener(topic = MqConstant.BILL_RECORD_TOPIC, consumerGroup = MqConstant.BILL_RECORD_CONSUME_GROUP)
//@RocketMQMessageListener(topic = MqConstant.BILL_RECORD_TOPIC, consumerGroup = MqConstant.BILL_RECORD_CONSUME_GROUP, consumeMode = ConsumeMode.ORDERLY)
public class BillMessageConsumer extends AbstractMessageListener<BillMessage> {
@Autowired

5
accompany-mq/accompany-mq-web/src/main/java/com/accompany/mq/consumer/GiftMessageConsumer.java
View File

@@ -5,7 +5,6 @@ import com.accompany.business.service.gift.GiftMessageService;
import com.accompany.mq.constant.MqConstant;
import com.accompany.mq.listener.AbstractMessageListener;
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;
@@ -14,8 +13,8 @@ import org.springframework.stereotype.Component;
@Slf4j
@Component
@ConditionalOnProperty(name = "spring.application.name", havingValue = "web")
//@RocketMQMessageListener(topic = MqConstant.GIFT_TOPIC, consumerGroup = MqConstant.GIFT_CONSUME_GROUP)
@RocketMQMessageListener(topic = MqConstant.GIFT_TOPIC, consumerGroup = MqConstant.GIFT_CONSUME_GROUP, consumeMode = ConsumeMode.ORDERLY)
@RocketMQMessageListener(topic = MqConstant.GIFT_TOPIC, consumerGroup = MqConstant.GIFT_CONSUME_GROUP)
//@RocketMQMessageListener(topic = MqConstant.GIFT_TOPIC, consumerGroup = MqConstant.GIFT_CONSUME_GROUP, consumeMode = ConsumeMode.ORDERLY)
public class GiftMessageConsumer extends AbstractMessageListener<GiftMessage> {
@Autowired