diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java index ba30d161..f80bd459 100644 --- a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/CommonBusinessRocketMqAutoConfiguration.java @@ -7,7 +7,11 @@ import org.springframework.context.annotation.ComponentScan; * @author pengliang on 2023/3/20 */ @Slf4j -@ComponentScan("com.red.circle.mq.rocket.business.producer") +@ComponentScan({ + "com.red.circle.mq.rocket.business.producer", + "com.red.circle.mq.rocket.business.config", + "com.red.circle.mq.rocket.business.service" +}) public class CommonBusinessRocketMqAutoConfiguration { } diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/config/MqThreadPoolConfig.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/config/MqThreadPoolConfig.java new file mode 100644 index 00000000..25bb3e3f --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/config/MqThreadPoolConfig.java @@ -0,0 +1,75 @@ +package com.red.circle.mq.rocket.business.config; + +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.RejectedExecutionHandler; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import lombok.extern.slf4j.Slf4j; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +/** + * MQ 消息发送线程池配置. + * + * @author Leo + */ +@Slf4j +@Configuration +public class MqThreadPoolConfig { + + @Bean(name = "mqCallbackExecutor", destroyMethod = "shutdown") + public ThreadPoolExecutor mqCallbackExecutor() { + + int corePoolSize = 50; // 核心线程 + int maximumPoolSize = 300; // 最大线程 + long keepAliveTime = 60L; + int queueCapacity = 10000; // 增大队列 + + ThreadPoolExecutor executor = new ThreadPoolExecutor( + corePoolSize, + maximumPoolSize, + keepAliveTime, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(queueCapacity), + new MqCallbackThreadFactory(), + new MqCallbackRejectedHandler() + ); + + executor.allowCoreThreadTimeOut(false); + + log.info("MQ 回调线程池初始化完成 - 核心线程:{}, 最大线程:{}, 队列容量:{}", + corePoolSize, maximumPoolSize, queueCapacity); + + return executor; + } + + private static class MqCallbackThreadFactory implements ThreadFactory { + + private final AtomicInteger threadNumber = new AtomicInteger(1); + private final String namePrefix = "mq-callback-"; + + @Override + public Thread newThread(Runnable r) { + Thread thread = new Thread(r, namePrefix + threadNumber.getAndIncrement()); + thread.setDaemon(false); + thread.setPriority(Thread.NORM_PRIORITY); + return thread; + } + } + + private static class MqCallbackRejectedHandler implements RejectedExecutionHandler { + + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + log.error("MQ 回调线程池队列已满！当前线程数:{}, 队列大小:{}, 任务被拒绝", + executor.getPoolSize(), + executor.getQueue().size()); + + if (!executor.isShutdown()) { + r.run(); + } + } + } +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/producer/GiftMqMessage.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/producer/GiftMqMessage.java index a5cc2bad..8eff3a34 100644 --- a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/producer/GiftMqMessage.java +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/producer/GiftMqMessage.java @@ -2,9 +2,9 @@ package com.red.circle.mq.rocket.business.producer; import com.red.circle.component.mq.MessageEvent; import com.red.circle.component.mq.service.FixedTopic; -import com.red.circle.component.mq.service.SenderService; import com.red.circle.mq.business.model.event.gift.GiveAwayGiftBatchEvent; import com.red.circle.mq.business.model.event.gift.OfflineProcessGiftEvent; +import com.red.circle.mq.rocket.business.service.MessageSenderService; import com.red.circle.mq.rocket.business.streams.GiveGiftSink; import com.red.circle.mq.rocket.business.streams.OfflineProcessGiftSink; import com.red.circle.tool.core.date.ZonedDateTimeAsiaRiyadhUtils; @@ -20,7 +20,7 @@ import org.springframework.stereotype.Component; @RequiredArgsConstructor public class GiftMqMessage { - private final SenderService senderService; + private final MessageSenderService senderService; /** diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/service/ManagedRocketMQProducer.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/service/ManagedRocketMQProducer.java new file mode 100644 index 00000000..70742435 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/service/ManagedRocketMQProducer.java @@ -0,0 +1,155 @@ +package com.red.circle.mq.rocket.business.service; + +import com.red.circle.component.mq.props.RocketMQProperties; +import com.red.circle.component.mq.service.*; +import com.red.circle.tool.core.collection.CollectionUtils; +import com.red.circle.tool.core.text.StringUtils; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ThreadPoolExecutor; +import javax.annotation.PostConstruct; +import javax.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.apis.ClientException; +import org.apache.rocketmq.client.apis.ClientServiceProvider; +import org.apache.rocketmq.client.apis.message.Message; +import org.apache.rocketmq.client.apis.message.MessageBuilder; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.stereotype.Component; + +/** + * 安全的 RocketMQ Producer. + * + * @author Leo + */ +@Slf4j +@Component +public class ManagedRocketMQProducer implements Producer { + + private RocketMQProperties rocketMQProperties; + private final Map producerMapping = new ConcurrentHashMap<>(); + private final ThreadPoolExecutor mqCallbackExecutor; + + public ManagedRocketMQProducer( + RocketMQProperties rocketMQProperties, + @Qualifier("mqCallbackExecutor") ThreadPoolExecutor mqCallbackExecutor) { + this.rocketMQProperties = rocketMQProperties; + this.mqCallbackExecutor = mqCallbackExecutor; + } + + private Message convertMessage(ProducerMessage producerMessage) { + if (Objects.isNull(producerMessage)) { + throw new IllegalArgumentException("producerMessage is null."); + } + + RocketMQProducerMapping mapping = getRocketMQProducerMapping(producerMessage.getTopic()); + MessageBuilder builder = mapping.getProvider() + .newMessageBuilder() + .setTopic(producerMessage.getTopic()) + .setKeys(producerMessage.getKeys()) + .setBody(producerMessage.getBody()); + + if (Objects.nonNull(producerMessage.getDeliveryTimestamp())) { + builder.setDeliveryTimestamp(producerMessage.getDeliveryTimestamp()); + } + + if (StringUtils.isNotBlank(producerMessage.getTag())) { + builder.setTag(producerMessage.getTag()); + } + + if (StringUtils.isNotBlank(producerMessage.getMessageGroup())) { + builder.setMessageGroup(producerMessage.getMessageGroup()); + } + + return builder.build(); + } + + private RocketMQProducerMapping getRocketMQProducerMapping(String topic) { + RocketMQProducerMapping mapping = producerMapping.get(topic); + if (Objects.isNull(mapping)) { + throw new IllegalArgumentException("Not Topic RocketMQProducerMapping."); + } + return mapping; + } + + @PostConstruct + public void start() { + List failedTopics = new ArrayList<>(); + + rocketMQProperties.getTopics().forEach((topic, value) -> { + try { + ClientServiceProvider provider = ClientServiceProvider.loadService(); + org.apache.rocketmq.client.apis.producer.Producer producer = provider + .newProducerBuilder() + .setClientConfiguration(rocketMQProperties.newClientConfiguration()) + .setTopics(new String[]{topic}) + .build(); + producerMapping.put(topic, new RocketMQProducerMapping(provider, producer)); + log.info("MQ Producer 初始化成功: topic={}", topic); + } catch (Throwable e) { + log.error("MQ Producer 初始化失败: topic={}", topic, e); + failedTopics.add(topic); + } + }); + + if (!failedTopics.isEmpty()) { + throw new RuntimeException("MQ Producer 初始化失败的 topic: " + failedTopics); + } + } + + public String send(ProducerMessage message) { + org.apache.rocketmq.client.apis.producer.Producer producer = + getRocketMQProducerMapping(message.getTopic()).getProducer(); + + if (Objects.isNull(producer)) { + throw new IllegalArgumentException("Topic=" + message.getTopic() + " config error."); + } + + try { + return producer.send(convertMessage(message)).getMessageId().toString(); + } catch (ClientException e) { + log.error("MQ 同步发送失败, topic:{}, tag:{}", + message.getTopic(), message.getTag(), e); + throw new RuntimeException("MQ send failed", e); + } + } + + + public void sendAsync(ProducerMessage message, SendCallback callback) { + org.apache.rocketmq.client.apis.producer.Producer producer = + getRocketMQProducerMapping(message.getTopic()).getProducer(); + + if (Objects.isNull(producer)) { + throw new IllegalArgumentException("Topic=" + message.getTopic() + " config error."); + } + + CompletableFuture future = + producer.sendAsync(convertMessage(message)); + + future.whenCompleteAsync((result, throwable) -> { + if (throwable != null) { + callback.onFail(throwable); + } else if (Objects.isNull(result.getMessageId())) { + callback.onFail(new RuntimeException("message Id is null.")); + } else { + callback.onSuccess(result.getMessageId().toString(), message); + } + }, mqCallbackExecutor); + } + + @PreDestroy + public void shutdown() { + if (CollectionUtils.isNotEmpty(producerMapping)) { + producerMapping.values().forEach(mapping -> { + mapping.close(); + log.info("producer [{}] close.", mapping.getProducer()); + }); + producerMapping.clear(); + } + } + +} diff --git a/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/service/MessageSenderService.java b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/service/MessageSenderService.java new file mode 100644 index 00000000..7e58aa92 --- /dev/null +++ b/rc-common-business/common-business-mq/business-rocketmq/src/main/java/com/red/circle/mq/rocket/business/service/MessageSenderService.java @@ -0,0 +1,104 @@ +package com.red.circle.mq.rocket.business.service; + +import com.red.circle.component.mq.MessageEvent; +import com.red.circle.component.mq.config.RocketMqFailMessage; +import com.red.circle.component.mq.config.RocketMqSenderFailService; +import com.red.circle.component.mq.service.FixedTopic; +import com.red.circle.component.mq.service.ProducerMessage; +import com.red.circle.component.mq.service.SendCallback; +import com.red.circle.tool.core.date.DateUtils; +import com.red.circle.tool.core.text.StringUtils; +import java.nio.charset.StandardCharsets; +import java.util.Objects; +import lombok.extern.slf4j.Slf4j; +import org.springframework.stereotype.Service; + +/** + * 安全的 MQ 发送服务. + * + * @author Leo + */ +@Slf4j +@Service +public class MessageSenderService { + + private final ManagedRocketMQProducer managedRocketMQProducer; + private final RocketMqSenderFailService rocketMqSenderFailService; + + public MessageSenderService( + ManagedRocketMQProducer managedRocketMQProducer, + RocketMqSenderFailService rocketMqSenderFailService) { + this.managedRocketMQProducer = managedRocketMQProducer; + this.rocketMqSenderFailService = rocketMqSenderFailService; + } + + public void sendEventMessage(final MessageEvent event) { + this.sendEventMessage(event, new SendCallback() { + @Override + public void onSuccess(String messageId, ProducerMessage message) { + + } + + @Override + public void onFail(Throwable throwable) { + log.error("SafeSenderService.sendEventMessage onFail: {},{}", + throwable.getMessage(), event.toJson()); + rocketMqSenderFailService.add(RocketMqFailMessage.builder() + .id(event.getConsumeId()) + .topic(event.getTopic()) + .tag(event.getTag()) + .errorMsg(throwable.getMessage()) + .body(event.getBody()) + .startDeliverTime(event.getStartDeliverTime()) + .process(Boolean.FALSE) + .createTime(DateUtils.now()) + .updateTime(DateUtils.now()) + .expiredTime(DateUtils.nowPlusDays(7)) + .build()); + } + }); + } + + public void sendEventMessage(MessageEvent event, SendCallback sendCallback) { + checkEvent(event); + checkSetDefaultTopic(event); + + ProducerMessage message = convertMessage(event); + managedRocketMQProducer.sendAsync(message, sendCallback); + } + + private void checkSetDefaultTopic(MessageEvent event) { + if (!StringUtils.isNotBlank(event.getTopic())) { + if (event.checkStartDeliverTimeMessage()) { + event.setTopic(FixedTopic.RC_DEFAULT_APP_DELAYED.getTopic()); + } else if (event.checkOrderMessage()) { + event.setTopic(FixedTopic.RC_DEFAULT_APP_ORDER.getTopic()); + } else { + event.setTopic(FixedTopic.RC_DEFAULT_APP_ORDINARY.getTopic()); + } + } + } + + private ProducerMessage convertMessage(MessageEvent event) { + return ProducerMessage.builder() + .topic(event.getTopic()) + .tag(event.getTag()) + .messageGroup(event.getMessageGroup()) + .keys(new String[]{event.messageKey()}) + .body(event.toBytes(StandardCharsets.UTF_8)) + .deliveryTimestamp(event.checkStartDeliverTimeMessage() ? event.getStartDeliverTime() : null) + .build(); + } + + private void checkEvent(MessageEvent event) { + if (Objects.isNull(event)) { + throw new IllegalArgumentException("SafeSenderService.sendObject event is null."); + } + if (StringUtils.isBlank(event.getTag())) { + throw new IllegalArgumentException("SafeSenderService.sendObject tag isBlank."); + } + if (StringUtils.isBlank(event.getBody())) { + throw new IllegalArgumentException("SafeSenderService.sendObject body isBlank."); + } + } +}