feat(mq-starter-rocketmq): rocket driver 发送(模板)+编程式消费(DefaultMQPushConsumer)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mq-starter-rocketmq/pom.xml

@@ -3,4 +3,12 @@
     <modelVersion>4.0.0</modelVersion>
     <parent><groupId>com.njcn</groupId><artifactId>mq-starter</artifactId><version>1.0.0-SNAPSHOT</version></parent>
     <artifactId>mq-starter-rocketmq</artifactId>
+    <dependencies>
+        <dependency><groupId>com.njcn</groupId><artifactId>mq-starter-core</artifactId><version>1.0.0-SNAPSHOT</version></dependency>
+        <dependency><groupId>org.apache.rocketmq</groupId><artifactId>rocketmq-spring-boot-starter</artifactId><version>2.2.3</version></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-autoconfigure</artifactId><version>${autoconfigure.version}</version></dependency>
+        <dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>${fastjson.version}</version></dependency>
+        <dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>${lombok.version}</version><optional>true</optional></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><version>${springboot.version}</version><scope>test</scope></dependency>
+    </dependencies>
 </project>

mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriver.java

@@ -0,0 +1,73 @@
+package com.njcn.mq.driver.rocketmq;
+
+import com.alibaba.fastjson.JSON;
+import com.njcn.mq.core.MqMessage;
+import com.njcn.mq.spi.MqDriver;
+import com.njcn.mq.spi.MqSubscription;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
+import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
+import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
+import org.apache.rocketmq.common.message.MessageExt;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.apache.rocketmq.spring.support.RocketMQHeaders;
+import org.springframework.messaging.support.MessageBuilder;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+
+@Slf4j
+public class RocketMqDriver implements MqDriver {
+
+    private final RocketMQTemplate template;
+    private final String namesrvAddr;
+    private final List<DefaultMQPushConsumer> consumers = new ArrayList<>();
+
+    public RocketMqDriver(RocketMQTemplate template, String namesrvAddr) {
+        this.template = template;
+        this.namesrvAddr = namesrvAddr;
+    }
+
+    @Override
+    public void send(String topic, MqMessage message) {
+        String tag = message.getTag();
+        String dest = (tag == null || tag.isEmpty()) ? topic : topic + ":" + tag;
+        String json = JSON.toJSONString(message.getPayload());
+        template.syncSend(dest, MessageBuilder.withPayload(json)
+                .setHeader(RocketMQHeaders.KEYS, message.getKey()).build());
+    }
+
+    @Override
+    public void subscribe(MqSubscription sub) {
+        try {
+            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(sub.getGroup());
+            consumer.setNamesrvAddr(namesrvAddr);
+            String tag = sub.getTag();
+            consumer.subscribe(sub.getTopic(), (tag == null || tag.isEmpty()) ? "*" : tag);
+            consumer.registerMessageListener((MessageListenerConcurrently) (msgs, context) -> {
+                for (MessageExt ext : msgs) {
+                    try {
+                        String json = new String(ext.getBody(), StandardCharsets.UTF_8);
+                        Object payload = JSON.parseObject(json, sub.getPayloadType());
+                        sub.getListener().onMessage(MqMessage.of(ext.getKeys(), ext.getTags(), payload));
+                    } catch (Exception e) {
+                        log.warn("[mq-rocket] handle failed msgId={}", ext.getMsgId(), e);
+                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
+                    }
+                }
+                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+            });
+            consumer.start();
+            consumers.add(consumer);
+        } catch (Exception e) {
+            throw new IllegalStateException("rocket subscribe failed: " + sub.getTopic(), e);
+        }
+    }
+
+    @Override
+    public void stop() {
+        for (DefaultMQPushConsumer c : consumers) c.shutdown();
+        consumers.clear();
+    }
+}

mq-starter-rocketmq/src/main/java/com/njcn/mq/driver/rocketmq/RocketMqDriverAutoConfiguration.java

@@ -0,0 +1,22 @@
+package com.njcn.mq.driver.rocketmq;
+
+import com.njcn.mq.spi.MqDriver;
+import org.apache.rocketmq.spring.autoconfigure.RocketMQProperties;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+@ConditionalOnClass(RocketMQTemplate.class)
+@ConditionalOnProperty(name = "mq.type", havingValue = "rocketmq", matchIfMissing = true)
+public class RocketMqDriverAutoConfiguration {
+
+    @Bean
+    @ConditionalOnMissingBean
+    public MqDriver mqDriver(RocketMQTemplate template, RocketMQProperties props) {
+        return new RocketMqDriver(template, props.getNameServer());
+    }
+}

mq-starter-rocketmq/src/main/resources/META-INF/spring.factories

@@ -0,0 +1,2 @@
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
+com.njcn.mq.driver.rocketmq.RocketMqDriverAutoConfiguration

mq-starter-rocketmq/src/test/java/com/njcn/mq/driver/rocketmq/RocketMqDriverSendTest.java

@@ -0,0 +1,31 @@
+package com.njcn.mq.driver.rocketmq;
+
+import com.njcn.mq.core.MqMessage;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.springframework.messaging.Message;
+
+import static org.mockito.Mockito.*;
+import static org.junit.jupiter.api.Assertions.*;
+
+class RocketMqDriverSendTest {
+    @Test
+    void send_withTag_buildsDestinationAndKeysHeader() {
+        RocketMQTemplate template = mock(RocketMQTemplate.class);
+        RocketMqDriver driver = new RocketMqDriver(template, "localhost:9876");
+        driver.send("File_Topic", MqMessage.of("k1", "node-7", "p"));
+        ArgumentCaptor<String> dest = ArgumentCaptor.forClass(String.class);
+        verify(template).syncSend(dest.capture(), any(Message.class));
+        assertEquals("File_Topic:node-7", dest.getValue());
+    }
+
+    @Test
+    void send_noTag_destinationIsTopicOnly() {
+        RocketMQTemplate template = mock(RocketMQTemplate.class);
+        new RocketMqDriver(template, "localhost:9876").send("LN_Topic", MqMessage.of("k", "", "p"));
+        ArgumentCaptor<String> dest = ArgumentCaptor.forClass(String.class);
+        verify(template).syncSend(dest.capture(), any(Message.class));
+        assertEquals("LN_Topic", dest.getValue());
+    }
+}