新增自定义发送mq接口，礼物mq重构

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 {
 
 }

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();
+      }
+    }
+  }
+}

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;
 
 
   /**

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<String, RocketMQProducerMapping> 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<String> 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<org.apache.rocketmq.client.apis.producer.SendReceipt> 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();
+    }
+  }
+
+}

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.");
+    }
+  }
+}