diff --git a/mq-spring-boot-starter/pom.xml b/mq-spring-boot-starter/pom.xml new file mode 100644 index 0000000..2f4d797 --- /dev/null +++ b/mq-spring-boot-starter/pom.xml @@ -0,0 +1,6 @@ + + 4.0.0 + com.njcnmq-starter1.0.0-SNAPSHOT + mq-spring-boot-starter + diff --git a/mq-starter-core/pom.xml b/mq-starter-core/pom.xml new file mode 100644 index 0000000..81c26d0 --- /dev/null +++ b/mq-starter-core/pom.xml @@ -0,0 +1,18 @@ + + 4.0.0 + com.njcnmq-starter1.0.0-SNAPSHOT + mq-starter-core + + org.springframework.bootspring-boot-autoconfigure${autoconfigure.version} + com.alibabafastjson${fastjson.version} + cn.hutoolhutool-all${hutool.version} + org.projectlomboklombok${lombok.version}true + org.springframework.bootspring-boot-configuration-processor${springboot.version}true + org.springframework.bootspring-boot-starter-test${springboot.version}test + + + org.apache.maven.pluginsmaven-compiler-plugin + org.apache.maven.pluginsmaven-surefire-plugin + + diff --git a/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java b/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java new file mode 100644 index 0000000..f81ebeb --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/annotation/MqListener.java @@ -0,0 +1,16 @@ +package com.njcn.mq.annotation; +import java.lang.annotation.ElementType; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; +import java.lang.annotation.Target; + +@Target(ElementType.METHOD) +@Retention(RetentionPolicy.RUNTIME) +public @interface MqListener { + String topic(); + String group(); + int concurrency() default 1; + int batchSize() default 1; + int maxRetry() default 3; + String tag() default "*"; +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/core/MqContext.java b/mq-starter-core/src/main/java/com/njcn/mq/core/MqContext.java new file mode 100644 index 0000000..28d83e7 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/core/MqContext.java @@ -0,0 +1,13 @@ +package com.njcn.mq.core; +import java.util.Map; + +/** @MqListener 方法可选第二参，只读暴露信封元数据。 */ +public class MqContext { + private final MqMessage message; + public MqContext(MqMessage message) { this.message = message; } + public String getKey() { return message.getKey(); } + public String getTag() { return message.getTag(); } + public String getSource() { return message.getSource(); } + public long getTimestamp() { return message.getTimestamp(); } + public Map getHeaders() { return message.getHeaders(); } +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/core/MqMessage.java b/mq-starter-core/src/main/java/com/njcn/mq/core/MqMessage.java new file mode 100644 index 0000000..5cebd2c --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/core/MqMessage.java @@ -0,0 +1,23 @@ +package com.njcn.mq.core; +import lombok.Builder; +import lombok.Data; +import java.util.HashMap; +import java.util.Map; + +@Data +@Builder +public class MqMessage { + private String key; + private String tag; + private String source; + private long timestamp; + private Map headers; + private Object payload; + + public static MqMessage of(String key, String tag, Object payload) { + return MqMessage.builder() + .key(key).tag(tag == null ? "" : tag).source("") + .timestamp(System.currentTimeMillis()) + .headers(new HashMap<>(4)).payload(payload).build(); + } +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqDriver.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqDriver.java new file mode 100644 index 0000000..7a34b04 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqDriver.java @@ -0,0 +1,8 @@ +package com.njcn.mq.spi; +import com.njcn.mq.core.MqMessage; + +public interface MqDriver { + void send(String topic, MqMessage message); + void subscribe(MqSubscription subscription); + void stop(); +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageListener.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageListener.java new file mode 100644 index 0000000..66e484c --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqMessageListener.java @@ -0,0 +1,6 @@ +package com.njcn.mq.spi; +import com.njcn.mq.core.MqMessage; + +public interface MqMessageListener { + void onMessage(MqMessage message) throws Exception; +} diff --git a/mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java new file mode 100644 index 0000000..b44f974 --- /dev/null +++ b/mq-starter-core/src/main/java/com/njcn/mq/spi/MqSubscription.java @@ -0,0 +1,16 @@ +package com.njcn.mq.spi; +import lombok.Builder; +import lombok.Data; + +@Data +@Builder +public class MqSubscription { + private String topic; + private String group; + private String tag; + private Class payloadType; + private int concurrency; + private int batchSize; + private int maxRetry; + private MqMessageListener listener; +} diff --git a/mq-starter-core/src/test/java/com/njcn/mq/core/MqMessageTest.java b/mq-starter-core/src/test/java/com/njcn/mq/core/MqMessageTest.java new file mode 100644 index 0000000..d19d6ad --- /dev/null +++ b/mq-starter-core/src/test/java/com/njcn/mq/core/MqMessageTest.java @@ -0,0 +1,16 @@ +package com.njcn.mq.core; +import org.junit.jupiter.api.Test; +import static org.junit.jupiter.api.Assertions.*; + +class MqMessageTest { + @Test + void of_setsKeyTagPayload_andContextReflectsThem() { + MqMessage m = MqMessage.of("k1", "nodeA", "hello"); + assertEquals("k1", m.getKey()); + assertEquals("nodeA", m.getTag()); + assertEquals("hello", m.getPayload()); + MqContext ctx = new MqContext(m); + assertEquals("k1", ctx.getKey()); + assertEquals("nodeA", ctx.getTag()); + } +} diff --git a/mq-starter-redis-stream/pom.xml b/mq-starter-redis-stream/pom.xml new file mode 100644 index 0000000..15f4a7e --- /dev/null +++ b/mq-starter-redis-stream/pom.xml @@ -0,0 +1,6 @@ + + 4.0.0 + com.njcnmq-starter1.0.0-SNAPSHOT + mq-starter-redis-stream + diff --git a/mq-starter-rocketmq/pom.xml b/mq-starter-rocketmq/pom.xml new file mode 100644 index 0000000..d95a801 --- /dev/null +++ b/mq-starter-rocketmq/pom.xml @@ -0,0 +1,6 @@ + + 4.0.0 + com.njcnmq-starter1.0.0-SNAPSHOT + mq-starter-rocketmq + diff --git a/pom.xml b/pom.xml new file mode 100644 index 0000000..0140fa9 --- /dev/null +++ b/pom.xml @@ -0,0 +1,38 @@ + + 4.0.0 + com.njcn + mq-starter + 1.0.0-SNAPSHOT + pom + mq-starter 统一 MQ 抽象层 + + mq-starter-core + mq-starter-rocketmq + mq-starter-redis-stream + mq-spring-boot-starter + + + nexus-releaseshttp://192.168.1.22:8001/nexus/content/repositories/releases/ + nexus-snapshotshttp://192.168.1.22:8001/nexus/content/repositories/snapshots/ + + + 8 + 8 + UTF-8 + 2.3.12.RELEASE + 2.3.12.RELEASE + 1.2.83 + 5.7.9 + 1.18.18 + + + + + org.apache.maven.pluginsmaven-compiler-plugin3.8.1 + 1.81.8UTF-8 + org.apache.maven.pluginsmaven-surefire-plugin2.22.2 + + + +