diff --git a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/ArtemisDemoController.java b/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/ArtemisDemoController.java index 893a4934e..1bc78488f 100644 --- a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/ArtemisDemoController.java +++ b/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/ArtemisDemoController.java @@ -11,6 +11,7 @@ import org.dromara.daxpay.payment.admin.controller.demo.param.SendDemoMessagePar import org.dromara.daxpay.payment.admin.controller.demo.result.DemoMessageResult; import org.dromara.daxpay.payment.admin.controller.demo.store.DemoMessageStore; import org.dromara.daxpay.platform.common.artemis.service.ArtemisTemplateService; +import org.dromara.daxpay.platform.common.json.util.JacksonUtil; import org.dromara.daxpay.platform.core.annotation.IgnoreAuth; import org.dromara.daxpay.platform.core.exception.BizInfoException; import org.dromara.daxpay.platform.core.rest.Res; @@ -58,23 +59,26 @@ public class ArtemisDemoController { .setScene(scene.name()) .setSendTime(OffsetDateTime.now()); + // 序列化为 JSON 字符串，发送层只负责搬运文本，不参与对象转换 + String json = JacksonUtil.toJson(message, false); + switch (scene) { case QUEUE -> { // 点对点：无 tag - artemisTemplateService.send(DemoArtemisConstants.QUEUE, null, message); + artemisTemplateService.send(DemoArtemisConstants.QUEUE, null, json); } case TOPIC -> { // 发布订阅：广播（必须走 sendTopic，否则 broker 端 multicast 地址会触发 ANYCAST 路由错误） - artemisTemplateService.sendTopic(DemoArtemisConstants.TOPIC, null, message); + artemisTemplateService.sendTopic(DemoArtemisConstants.TOPIC, null, json); } case DELAY -> { // 延时：调用 sendDelay（已校验 delaySeconds 非空） artemisTemplateService.sendDelay( - DemoArtemisConstants.DELAY_QUEUE, null, message, param.getDelaySeconds()); + DemoArtemisConstants.DELAY_QUEUE, null, json, param.getDelaySeconds()); } case TAG -> { // Tag 过滤：把标签作为消息属性写入（已校验 tag 非空） - artemisTemplateService.send(DemoArtemisConstants.TAG_QUEUE, param.getTag(), message); + artemisTemplateService.send(DemoArtemisConstants.TAG_QUEUE, param.getTag(), json); } } return Res.ok(); diff --git a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/config/DemoJmsListenerConfig.java b/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/config/DemoJmsListenerConfig.java deleted file mode 100644 index e09e9b158..000000000 --- a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/config/DemoJmsListenerConfig.java +++ /dev/null @@ -1,35 +0,0 @@ -package org.dromara.daxpay.payment.admin.controller.demo.config; - -import jakarta.jms.ConnectionFactory; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.jms.config.DefaultJmsListenerContainerFactory; -import org.springframework.jms.support.converter.MessageConverter; - -/// # Artemis 演示 JMS 监听容器配置 -/// -/// 为演示 Topic 消费者提供独立的 `JmsListenerContainerFactory`，唯一职责是开启 Topic 广播语义。 -/// -/// 说明： -/// - `@EnableJms` 已由缓存模块 `CacheJmsListenerConfig` 全局开启，此处无需重复声明 -/// - Queue 消费者使用默认 `jmsListenerContainerFactory`（pub-sub-domain=false） -/// - Topic 消费者必须显式指定本工厂 -@Configuration -public class DemoJmsListenerConfig { - - /// 演示 Topic 监听容器工厂（pub-sub 模式） - @Bean(name = "demoTopicListenerFactory") - public DefaultJmsListenerContainerFactory demoTopicListenerFactory( - ConnectionFactory connectionFactory, - MessageConverter messageConverter) { - - DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); - factory.setConnectionFactory(connectionFactory); - // 复用 Artemis 统一的 JSON 转换器 - factory.setMessageConverter(messageConverter); - // Topic 模式（pub-sub），对应 broker 端 multicast 路由类型 - factory.setPubSubDomain(true); - factory.setAutoStartup(true); - return factory; - } -} diff --git a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/consumer/DemoTopicConsumer.java b/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/consumer/DemoTopicConsumer.java index 8de0c6705..b929f73da 100644 --- a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/consumer/DemoTopicConsumer.java +++ b/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/consumer/DemoTopicConsumer.java @@ -15,7 +15,7 @@ import org.springframework.stereotype.Component; /// 监听 `demo.topic`，演示 JMS pub-sub 广播场景。 /// 同一节点注册两个订阅者，证明发布订阅模式下每个订阅者都能收到完整消息。 /// -/// 注意：必须显式指定 `containerFactory = "demoTopicListenerFactory"`， +/// 注意：必须显式指定 `containerFactory = "topicListenerFactory"`， /// 否则走默认 Queue 工厂，Topic 消息将无法被消费。 @Slf4j @Component @@ -25,13 +25,13 @@ public class DemoTopicConsumer { private final DemoMessageStore store; /// 订阅者 A - @JmsListener(destination = DemoArtemisConstants.TOPIC, containerFactory = "demoTopicListenerFactory") + @JmsListener(destination = DemoArtemisConstants.TOPIC, containerFactory = "topicListenerFactory") public void onMessageA(String json) { handle(json, "demo-topic-consumer-A"); } /// 订阅者 B（同节点模拟多订阅者） - @JmsListener(destination = DemoArtemisConstants.TOPIC, containerFactory = "demoTopicListenerFactory") + @JmsListener(destination = DemoArtemisConstants.TOPIC, containerFactory = "topicListenerFactory") public void onMessageB(String json) { handle(json, "demo-topic-consumer-B"); } diff --git a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/message/DemoArtemisMessage.java b/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/message/DemoArtemisMessage.java index 10b2727b4..b0c3a26f5 100644 --- a/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/message/DemoArtemisMessage.java +++ b/daxpay-payment/daxpay-payment-admin/src/main/java/org/dromara/daxpay/payment/admin/controller/demo/message/DemoArtemisMessage.java @@ -12,7 +12,8 @@ import java.time.OffsetDateTime; /// 通过 JMS 在生产者与消费者之间传递的消息载体。 /// 必须保留无参构造，否则 Jackson 反序列化会失败。 /// -/// @see org.dromara.daxpay.platform.common.artemis.message.ArtemisMessageConverter +/// 传输时由生产端用 `JacksonUtil.toJson` 序列化为 JSON 字符串， +/// 消费端 `onMessage(String json)` 拿到文本后自行 `JacksonUtil.toBean` 反序列化。 @Data @NoArgsConstructor @AllArgsConstructor diff --git a/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/configuration/CacheJmsListenerConfig.java b/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/configuration/CacheJmsListenerConfig.java deleted file mode 100644 index 743a57d1d..000000000 --- a/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/configuration/CacheJmsListenerConfig.java +++ /dev/null @@ -1,42 +0,0 @@ -package org.dromara.daxpay.platform.capability.cache.configuration; - -import org.dromara.daxpay.platform.common.artemis.message.ArtemisMessageConverter; -import jakarta.jms.ConnectionFactory; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.jms.annotation.EnableJms; -import org.springframework.jms.config.DefaultJmsListenerContainerFactory; -import org.springframework.jms.support.converter.MessageConverter; - -/// # 缓存失效 JMS 监听容器配置 -/// -/// 为缓存失效消费者提供独立的 `JmsListenerContainerFactory`，唯一职责是开启 Topic 广播语义。 -/// -/// 为什么需要独立工厂： -/// - Spring Boot 默认的 `jmsListenerContainerFactory` 是 Queue 模式（pub-sub-domain=false） -/// - `pubSubDomain` 是工厂级配置，无法在 `@JmsListener` 注解上覆盖 -/// - 缓存失效必须是 Topic 广播，否则消息只被一个节点消费，导致其他节点 L1 缓存不一致 -/// -/// 设计取舍： -/// - 不用 durable subscription：缓存失效消息可丢，节点重启后 L1 会重建，有 TTL 兜底无需补收 -/// - 不设 clientId：non-durable 订阅不强制 clientId 唯一，省去 hostname+PID 拼接的出错面 -@Configuration -@EnableJms -public class CacheJmsListenerConfig { - - /// 缓存失效 Topic 监听容器工厂 - @Bean(name = "cacheTopicListenerFactory") - public DefaultJmsListenerContainerFactory cacheTopicListenerFactory( - ConnectionFactory connectionFactory, - MessageConverter messageConverter) { - - DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); - factory.setConnectionFactory(connectionFactory); - // 使用 Artemis 统一的 JSON 转换器（@ConditionalOnMissingBean 保证此处注入的是 ArtemisMessageConverter） - factory.setMessageConverter(messageConverter); - // Topic 模式（pub-sub），对应 broker 端 multicast 路由类型 —— 本工厂存在的唯一理由 - factory.setPubSubDomain(true); - factory.setAutoStartup(true); - return factory; - } -} diff --git a/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/consumer/CacheInvalidationConsumer.java b/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/consumer/CacheInvalidationConsumer.java index f78059bcf..42b3d2c99 100644 --- a/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/consumer/CacheInvalidationConsumer.java +++ b/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/consumer/CacheInvalidationConsumer.java @@ -35,12 +35,12 @@ public class CacheInvalidationConsumer { /// 订阅缓存失效 Topic /// - /// 通过独立的 listenerContainerFactory（`cacheTopicListenerFactory`）配置 Topic 模式， + /// 通过通用的 `topicListenerFactory`（pubSubDomain=true）配置 Topic 模式， /// 每个 non-durable 订阅者都会收到消息，实现跨节点广播。 /// 方法签名接收原始 JSON 字符串（统一 Text 传输），手动反序列化为目标类型。 @JmsListener( destination = CacheTopicConstants.TOPIC, - containerFactory = "cacheTopicListenerFactory" + containerFactory = "topicListenerFactory" ) public void onMessage(String json) { CacheInvalidationMessage message; diff --git a/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/publisher/CacheInvalidationPublisher.java b/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/publisher/CacheInvalidationPublisher.java index 950f706f2..fb8bc80ff 100644 --- a/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/publisher/CacheInvalidationPublisher.java +++ b/daxpay-platform/daxpay-platform-capability/capability-cache/src/main/java/org/dromara/daxpay/platform/capability/cache/notify/publisher/CacheInvalidationPublisher.java @@ -4,6 +4,7 @@ import org.dromara.daxpay.platform.capability.cache.notify.message.CacheInvalida import org.dromara.daxpay.platform.capability.cache.notify.message.CacheInvalidationType; import org.dromara.daxpay.platform.capability.cache.notify.support.CacheTopicConstants; import org.dromara.daxpay.platform.common.artemis.service.ArtemisTemplateService; +import org.dromara.daxpay.platform.common.json.util.JacksonUtil; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -36,7 +37,8 @@ public class CacheInvalidationPublisher { try { // 缓存失效是广播语义，必须走 sendTopic（broker 端 cache-invalidation-topic 为 multicast 路由） - artemisTemplateService.sendTopic(CacheTopicConstants.TOPIC, CacheTopicConstants.TAG_EVICT, message); + artemisTemplateService.sendTopic( + CacheTopicConstants.TOPIC, CacheTopicConstants.TAG_EVICT, JacksonUtil.toJson(message, false)); log.debug("发布缓存失效消息成功: cacheName={}, key={}", cacheName, key); } catch (Exception e) { log.error("发布缓存失效消息失败: cacheName={}, key={}, error={}", cacheName, key, e.getMessage(), e); @@ -53,7 +55,8 @@ public class CacheInvalidationPublisher { try { // 缓存清空也是广播语义，同样使用 sendTopic - artemisTemplateService.sendTopic(CacheTopicConstants.TOPIC, CacheTopicConstants.TAG_CLEAR, message); + artemisTemplateService.sendTopic( + CacheTopicConstants.TOPIC, CacheTopicConstants.TAG_CLEAR, JacksonUtil.toJson(message, false)); log.debug("发布缓存清空消息成功: cacheName={}", cacheName); } catch (Exception e) { log.error("发布缓存清空消息失败: cacheName={}, error={}", cacheName, e.getMessage(), e); diff --git a/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/ArtemisCommonAutoConfiguration.java b/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/ArtemisCommonAutoConfiguration.java index 17ee09347..1399e0456 100644 --- a/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/ArtemisCommonAutoConfiguration.java +++ b/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/ArtemisCommonAutoConfiguration.java @@ -1,14 +1,15 @@ package org.dromara.daxpay.platform.common.artemis; import jakarta.jms.ConnectionFactory; -import org.dromara.daxpay.platform.common.artemis.message.ArtemisMessageConverter; import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.jms.autoconfigure.JmsAutoConfiguration; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.ComponentScan; +import org.springframework.jms.annotation.EnableJms; +import org.springframework.jms.config.DefaultJmsListenerContainerFactory; +import org.springframework.jms.config.JmsListenerContainerFactory; import org.springframework.jms.core.JmsTemplate; -import org.springframework.jms.support.converter.MessageConverter; /// # Artemis 通用模块自动配置入口 /// @@ -16,50 +17,67 @@ import org.springframework.jms.support.converter.MessageConverter; /// /// 注册内容： /// - 组件扫描加载 {@link org.dromara.daxpay.platform.common.artemis.service.ArtemisTemplateService} -/// - 注册 {@link ArtemisMessageConverter} 为默认 {@link MessageConverter}， -/// 使 JmsTemplate / JmsClient 的 `send(Object payload)` 自动序列化为 JSON TextMessage -/// - 注册独立的 `topicJmsTemplate`（pubSubDomain=true），供 Topic 广播场景使用 +/// - `@EnableJms`：全局开启 `@JmsListener` 注解处理 +/// - `topicJmsTemplate`（pubSubDomain=true）：发送端 Topic 能力 +/// - `topicListenerFactory`（pubSubDomain=true）：消费端 Topic 能力 +/// +/// 为什么 Queue / Topic 必须分两套 Bean： +/// - `pubSubDomain` 是 JmsTemplate / JmsListenerContainerFactory 的实例级配置， +/// 决定 destination 解析为 Queue（anycast）还是 Topic（multicast） +/// - Spring Boot 默认装配的 `jmsTemplate` / `jmsListenerContainerFactory` 是 Queue 模式 +/// （由 `spring.jms.pub-sub-domain=false` 决定） +/// - Topic 场景必须用本类提供的 topic 版本，否则 broker 端路由类型不匹配 +/// +/// 消息传输约定： +/// - 消息体统一为 JSON 字符串，由调用方自行序列化/反序列化 +/// - 不注册 MessageConverter，回落到 Spring 默认的 `SimpleMessageConverter` +/// —— 对 String payload 直接 `createTextMessage(string)`，对 `onMessage(String)` 直接返回文本 /// /// 约束： /// - 必须 `after = JmsAutoConfiguration.class`，否则我们的 `topicJmsTemplate`（也是 JmsTemplate 类型） /// 会让 Spring Boot 内部的 `@ConditionalOnMissingBean(JmsTemplate.class)` 不通过， /// 导致默认 `jmsTemplate` Bean 不被创建，业务方注入时报 `No qualifying bean ... @Qualifier("jmsTemplate")` /// - broker 端需保证相关 address 存在；若开启 `auto-create-jms-queues/addresses`，则发送时自动创建 -/// - 广播场景（对应 RocketMQ BROADCASTING）要求 broker 端把 address 配为 multicast 路由类型， -/// 并在客户端用独立 durable subscription 名 +/// - 广播场景（对应 RocketMQ BROADCASTING）要求 broker 端把 address 配为 multicast 路由类型 @AutoConfiguration(after = JmsAutoConfiguration.class) @ComponentScan +@EnableJms public class ArtemisCommonAutoConfiguration { - /// Topic 模板的 Bean 名，避免与 Spring Boot 默认的 `jmsTemplate` 冲突 + /// Topic 发送模板的 Bean 名 public static final String TOPIC_JMS_TEMPLATE = "topicJmsTemplate"; - /// 默认消息转换器：统一使用 JSON + TextMessage - /// - /// 仅当容器中不存在其他 MessageConverter 时生效，避免覆盖业务自定义实现 - @Bean - @ConditionalOnMissingBean(MessageConverter.class) - public ArtemisMessageConverter artemisMessageConverter() { - return new ArtemisMessageConverter(); - } + /// Topic 监听容器工厂的 Bean 名，供所有 `@JmsListener` 订阅 Topic 时引用 + public static final String TOPIC_LISTENER_FACTORY = "topicListenerFactory"; - /// Topic 专用 JmsTemplate（pubSubDomain=true） + /// Topic 发送专用 JmsTemplate（pubSubDomain=true） /// - /// 为什么需要独立 Bean： - /// - Spring Boot 自动装配的 `jmsTemplate` 由 `spring.jms.pub-sub-domain` 决定 destination 类型 - /// - 项目默认 `pub-sub-domain=false`（点对点 Queue），发到 Topic 地址会触发 Artemis broker - /// `Destination ... does not support ANYCAST routing` 异常 - /// - `pubSubDomain` 是 JmsTemplate 实例级配置，无法在调用时切换，必须独立 Bean + /// 与 Spring Boot 默认的 `jmsTemplate`（Queue 模式）共存， + /// 由 `ArtemisTemplateService` 通过 `@Qualifier` 注入。 /// - /// `@ConditionalOnMissingBean(name = TOPIC_JMS_TEMPLATE)`：仅当业务方未自定义同名 Bean 时生效。 - /// 复用统一的 {@link MessageConverter}，序列化行为与默认模板保持一致。 + /// 不设置 MessageConverter，回落到 Spring 默认 `SimpleMessageConverter`（String ↔ TextMessage 透传）。 @Bean(TOPIC_JMS_TEMPLATE) @ConditionalOnMissingBean(name = TOPIC_JMS_TEMPLATE) - public JmsTemplate topicJmsTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) { + public JmsTemplate topicJmsTemplate(ConnectionFactory connectionFactory) { JmsTemplate template = new JmsTemplate(connectionFactory); // 关键：开启 pub-sub 模式，destination 解析为 Topic（multicast） template.setPubSubDomain(true); - template.setMessageConverter(messageConverter); return template; } + + /// Topic 消费专用监听容器工厂（pubSubDomain=true） + /// + /// 所有需要订阅 Topic 的 `@JmsListener` 通过 + /// `containerFactory = "topicListenerFactory"` 引用， + /// 取代各业务模块自行创建重复的 Topic ListenerFactory。 + @Bean(TOPIC_LISTENER_FACTORY) + @ConditionalOnMissingBean(name = TOPIC_LISTENER_FACTORY) + public JmsListenerContainerFactory topicListenerFactory(ConnectionFactory connectionFactory) { + DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory(); + factory.setConnectionFactory(connectionFactory); + // Topic 模式（pub-sub），对应 broker 端 multicast 路由类型 + factory.setPubSubDomain(true); + factory.setAutoStartup(true); + return factory; + } } diff --git a/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/message/ArtemisMessageConverter.java b/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/message/ArtemisMessageConverter.java deleted file mode 100644 index 3d3aef125..000000000 --- a/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/message/ArtemisMessageConverter.java +++ /dev/null @@ -1,56 +0,0 @@ -package org.dromara.daxpay.platform.common.artemis.message; - -import org.dromara.daxpay.platform.common.artemis.exception.ArtemisException; -import org.dromara.daxpay.platform.common.json.util.JacksonUtil; -import jakarta.jms.JMSException; -import jakarta.jms.Message; -import jakarta.jms.Session; -import jakarta.jms.TextMessage; -import lombok.extern.slf4j.Slf4j; -import org.springframework.jms.support.converter.MessageConversionException; -import org.springframework.jms.support.converter.MessageConverter; - -/// # Artemis 消息转换器 -/// -/// 统一使用 TextMessage + JSON 文本承载消息，实现「纯 Text 传输」。 -/// -/// 设计要点： -/// - 发送端任意对象统一由 Jackson 序列化为 JSON 字符串，写入 TextMessage -/// - 不再向消息属性写入 Java 类型信息（如 `_type`），避免生产/消费两端类型绑定 -/// - 消费端 `@JmsListener` 方法签名统一为 `onMessage(String json)`，自行调用 `JacksonUtil.toBean` 反序列化 -/// - 这样消息保持自描述的 JSON 契约，类改名、跨服务消费、多语言接入都安全 -/// -/// @see MessageConverter Spring JMS 标准消息转换接口 -@Slf4j -public class ArtemisMessageConverter implements MessageConverter { - - /// 序列化目标：使用 TextMessage 承载 JSON - @Override - public Message toMessage(Object object, Session session) throws JMSException, MessageConversionException { - String json; - try { - // 非格式化 JSON，节省传输体积 - json = JacksonUtil.toJson(object, false); - } catch (Exception e) { - log.error("Artemis 消息序列化失败: {}", e.getMessage(), e); - throw new ArtemisException("error.artemis.serializeFailed", e.getMessage()); - } - - // 仅写入 JSON 文本，不携带任何类型属性，保持消息与 Java 类型解耦 - return session.createTextMessage(json); - } - - /// 反序列化：直接返回 TextMessage 的文本内容 - /// - /// 永远返回 String（JSON 文本），由消费端按需手动反序列化为目标类型。 - /// 因此 `@JmsListener` 方法签名统一使用 `onMessage(String json)`。 - @Override - public Object fromMessage(Message message) throws JMSException, MessageConversionException { - if (!(message instanceof TextMessage textMessage)) { - log.error("Artemis 消息体类型不支持，期望 TextMessage，实际: {}", message.getClass().getName()); - throw new ArtemisException("error.artemis.parseBodyFailed", - "expected TextMessage but got " + message.getClass().getName()); - } - return textMessage.getText(); - } -} diff --git a/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/service/ArtemisTemplateService.java b/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/service/ArtemisTemplateService.java index 6a3427d32..b4b7b9240 100644 --- a/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/service/ArtemisTemplateService.java +++ b/daxpay-platform/daxpay-platform-common/common-artemis/src/main/java/org/dromara/daxpay/platform/common/artemis/service/ArtemisTemplateService.java @@ -33,6 +33,13 @@ import java.util.Map; /// - 错配会触发 broker 端 `Destination ... does not support ANYCAST/MULTICAST routing` 异常 /// - 调用方在编码期就明确语义，避免运行期歧义 /// +/// 消息体传输约定： +/// - `body` 统一为 JSON 字符串，**由调用方自行序列化**（推荐 `JacksonUtil.toJson(obj)`） +/// - 发送端不参与对象转换，回落到 Spring 默认 `SimpleMessageConverter`： +/// `String` payload 直接写入 `TextMessage` +/// - 消费端 `@JmsListener` 方法签名统一 `onMessage(String json)`，自行反序列化 +/// - 这样消息保持自描述的 JSON 契约，类改名、跨服务消费、多语言接入都安全 +/// /// @see JmsClient Spring Framework 7 fluent JMS 客户端 @Slf4j @Service @@ -62,8 +69,8 @@ public class ArtemisTemplateService { /// /// @param address 目标地址（对应 Artemis address，kebab-case 命名） /// @param tag 消息标签，用于消息过滤，为空时不设置 - /// @param body 消息体，由 MessageConverter 自动序列化为 JSON TextMessage - public void send(String address, String tag, Object body) { + /// @param body JSON 字符串消息体，由调用方自行序列化 + public void send(String address, String tag, String body) { Map headers = buildHeaders(tag); try { jmsClient(queueJmsTemplate).destination(address).send(body, headers); @@ -82,9 +89,9 @@ public class ArtemisTemplateService { /// /// @param address 目标地址 /// @param tag 消息标签 - /// @param body 消息体 + /// @param body JSON 字符串消息体，由调用方自行序列化 /// @param delaySeconds 延时秒数 - public void sendDelay(String address, String tag, Object body, int delaySeconds) { + public void sendDelay(String address, String tag, String body, int delaySeconds) { Map headers = buildHeaders(tag); long delayMillis = delaySeconds * 1000L; try { @@ -107,9 +114,9 @@ public class ArtemisTemplateService { /// /// @param address 目标地址 /// @param tag 消息标签 - /// @param body 消息体 + /// @param body JSON 字符串消息体，由调用方自行序列化 /// @param deliveryTime 投递时刻（统一使用 OffsetDateTime，避免时区问题） - public void sendDelayAt(String address, String tag, Object body, OffsetDateTime deliveryTime) { + public void sendDelayAt(String address, String tag, String body, OffsetDateTime deliveryTime) { Map headers = buildHeaders(tag); long deliveryTimestamp = deliveryTime.toInstant().toEpochMilli(); long delayMillis = deliveryTimestamp - System.currentTimeMillis(); @@ -139,8 +146,8 @@ public class ArtemisTemplateService { /// /// @param address 目标 Topic 地址 /// @param tag 消息标签 - /// @param body 消息体 - public void sendTopic(String address, String tag, Object body) { + /// @param body JSON 字符串消息体，由调用方自行序列化 + public void sendTopic(String address, String tag, String body) { Map headers = buildHeaders(tag); try { jmsClient(topicJmsTemplate).destination(address).send(body, headers); @@ -156,9 +163,9 @@ public class ArtemisTemplateService { /// /// @param address 目标 Topic 地址 /// @param tag 消息标签 - /// @param body 消息体 + /// @param body JSON 字符串消息体，由调用方自行序列化 /// @param delaySeconds 延时秒数 - public void sendTopicDelay(String address, String tag, Object body, int delaySeconds) { + public void sendTopicDelay(String address, String tag, String body, int delaySeconds) { Map headers = buildHeaders(tag); long delayMillis = delaySeconds * 1000L; try { @@ -178,9 +185,9 @@ public class ArtemisTemplateService { /// /// @param address 目标 Topic 地址 /// @param tag 消息标签 - /// @param body 消息体 + /// @param body JSON 字符串消息体，由调用方自行序列化 /// @param deliveryTime 投递时刻 - public void sendTopicDelayAt(String address, String tag, Object body, OffsetDateTime deliveryTime) { + public void sendTopicDelayAt(String address, String tag, String body, OffsetDateTime deliveryTime) { Map headers = buildHeaders(tag); long deliveryTimestamp = deliveryTime.toInstant().toEpochMilli(); long delayMillis = deliveryTimestamp - System.currentTimeMillis();