mirror of
https://gitee.com/dromara/dax-pay
synced 2026-08-10 23:01:58 +08:00
refactor(mq): 移除 ArtemisTemplateService 的 tag 参数及全部 TAG 相关代码
tag（JMS selector）在当前所有场景中均无实际价值： - Demo TAG 场景用固定值 important/normal，不同 address 更清晰 - 缓存失效场景消息体已有 type 字段，tag 写了无 selector 消费 - 延时任务是低频单消费者，selector 无用武之地 改动： - ArtemisTemplateService 6 个方法签名去掉 tag 参数，buildHeaders 简化 - 删除 DemoTagConsumer 及 DemoArtemisConstants 的 TAG 常量 - SendDemoMessageParam/DemoArtemisMessage/DemoMessageResult 删除 tag 字段 - ArtemisDemoController 删除 TAG 场景分支及校验 - CacheTopicConstants/CacheInvalidationPublisher 删除 TAG_EVICT/TAG_CLEAR - error/demo.json 删除 tagRequired 国际化
This commit is contained in:
DaxPay Dev
@@ -1,7 +1,6 @@
|
||||
package org.dromara.daxpay.payment.admin.controller.demo;
|
||||
|
||||
import cn.hutool.core.lang.UUID;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import io.swagger.v3.oas.annotations.Operation;
|
||||
import io.swagger.v3.oas.annotations.tags.Tag;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
@@ -28,7 +27,7 @@ import java.util.List;
|
||||
|
||||
/// # Artemis 消息队列演示接口
|
||||
///
|
||||
/// 演示 JMS 四类核心场景：点对点队列、发布订阅、延时消息、Tag 过滤。
|
||||
/// 演示 JMS 三类核心场景：点对点队列、发布订阅、延时消息。
|
||||
/// 消费记录暂存于内存，前端通过 `/list` 轮询拉取。
|
||||
///
|
||||
/// 鉴权：URL 前缀 `/demo/**` 已在白名单，类上叠加 `@IgnoreAuth` 双保险。
|
||||
@@ -55,7 +54,6 @@ public class ArtemisDemoController {
|
||||
DemoArtemisMessage message = new DemoArtemisMessage()
|
||||
.setId(UUID.randomUUID().toString(true))
|
||||
.setContent(param.getContent())
|
||||
.setTag(param.getTag())
|
||||
.setScene(scene.name())
|
||||
.setSendTime(OffsetDateTime.now());
|
||||
|
||||
@@ -64,21 +62,17 @@ public class ArtemisDemoController {
|
||||
|
||||
switch (scene) {
|
||||
case QUEUE -> {
|
||||
// 点对点：无 tag
|
||||
artemisTemplateService.send(DemoArtemisConstants.QUEUE, null, json);
|
||||
// 点对点
|
||||
artemisTemplateService.send(DemoArtemisConstants.QUEUE, json);
|
||||
}
|
||||
case TOPIC -> {
|
||||
// 发布订阅：广播（必须走 sendTopic，否则 broker 端 multicast 地址会触发 ANYCAST 路由错误）
|
||||
artemisTemplateService.sendTopic(DemoArtemisConstants.TOPIC, null, json);
|
||||
artemisTemplateService.sendTopic(DemoArtemisConstants.TOPIC, json);
|
||||
}
|
||||
case DELAY -> {
|
||||
// 延时：调用 sendDelay（已校验 delaySeconds 非空）
|
||||
artemisTemplateService.sendDelay(
|
||||
DemoArtemisConstants.DELAY_QUEUE, null, json, param.getDelaySeconds());
|
||||
}
|
||||
case TAG -> {
|
||||
// Tag 过滤：把标签作为消息属性写入（已校验 tag 非空）
|
||||
artemisTemplateService.send(DemoArtemisConstants.TAG_QUEUE, param.getTag(), json);
|
||||
DemoArtemisConstants.DELAY_QUEUE, json, param.getDelaySeconds());
|
||||
}
|
||||
}
|
||||
return Res.ok();
|
||||
@@ -101,9 +95,6 @@ public class ArtemisDemoController {
|
||||
|
||||
/// 校验与场景绑定的必填字段
|
||||
private void validateSceneParam(SendDemoMessageParam param, SendDemoMessageParam.SendScene scene) {
|
||||
if (scene == SendDemoMessageParam.SendScene.TAG && StrUtil.isEmpty(param.getTag())) {
|
||||
throw new BizInfoException("error.demo.tagRequired");
|
||||
}
|
||||
if (scene == SendDemoMessageParam.SendScene.DELAY && param.getDelaySeconds() == null) {
|
||||
throw new BizInfoException("error.demo.delaySecondsRequired");
|
||||
}
|
||||
|
||||
@@ -2,7 +2,7 @@ package org.dromara.daxpay.payment.admin.controller.demo.constant;
|
||||
|
||||
/// # Artemis 消息队列演示常量
|
||||
///
|
||||
/// 演示用的 address 与 tag 常量，地址命名遵循 kebab-case 约定。
|
||||
/// 演示用的 address 常量，地址命名遵循 kebab-case 约定。
|
||||
/// Artemis 默认开启地址自动创建，无需在 broker 端预置。
|
||||
///
|
||||
/// @see org.dromara.daxpay.platform.common.artemis.service.ArtemisTemplateService
|
||||
@@ -16,13 +16,4 @@ public interface DemoArtemisConstants {
|
||||
|
||||
/// 延时消息队列 address
|
||||
String DELAY_QUEUE = "demo.delay";
|
||||
|
||||
/// Tag 过滤演示队列 address
|
||||
String TAG_QUEUE = "demo.tag";
|
||||
|
||||
/// 重要消息 Tag
|
||||
String TAG_IMPORTANT = "important";
|
||||
|
||||
/// 普通消息 Tag
|
||||
String TAG_NORMAL = "normal";
|
||||
}
|
||||
|
||||
@@ -1,54 +0,0 @@
|
||||
package org.dromara.daxpay.payment.admin.controller.demo.consumer;
|
||||
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.daxpay.payment.admin.controller.demo.constant.DemoArtemisConstants;
|
||||
import org.dromara.daxpay.payment.admin.controller.demo.message.DemoArtemisMessage;
|
||||
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.json.util.JacksonUtil;
|
||||
import org.springframework.jms.annotation.JmsListener;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/// # Tag 过滤消费者（演示）
|
||||
///
|
||||
/// 监听 `demo.tag`，演示通过 JMS selector 按消息属性过滤消费。
|
||||
/// 两个监听方法订阅同一地址，但分别只消费 `tag = 'important'` 和 `tag = 'normal'` 的消息。
|
||||
///
|
||||
/// 工作原理：
|
||||
/// - 生产者发送时在消息属性中写入 `tag`（由 `ArtemisTemplateService.HEADER_TAG = "tag"` 设置）
|
||||
/// - selector 表达式基于消息属性匹配，`tag` 为字符串需用单引号包裹
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
public class DemoTagConsumer {
|
||||
|
||||
private final DemoMessageStore store;
|
||||
|
||||
/// 只消费 important 标签的消息
|
||||
@JmsListener(destination = DemoArtemisConstants.TAG_QUEUE, selector = "tag = '" + DemoArtemisConstants.TAG_IMPORTANT + "'")
|
||||
public void onImportant(String json) {
|
||||
handle(json, "demo-tag-consumer-important");
|
||||
}
|
||||
|
||||
/// 只消费 normal 标签的消息
|
||||
@JmsListener(destination = DemoArtemisConstants.TAG_QUEUE, selector = "tag = '" + DemoArtemisConstants.TAG_NORMAL + "'")
|
||||
public void onNormal(String json) {
|
||||
handle(json, "demo-tag-consumer-normal");
|
||||
}
|
||||
|
||||
private void handle(String json, String consumer) {
|
||||
DemoArtemisMessage message;
|
||||
try {
|
||||
// 统一 Text 传输，消费端手动反序列化为目标类型
|
||||
message = JacksonUtil.toBean(json, DemoArtemisMessage.class);
|
||||
} catch (Exception e) {
|
||||
log.warn("Tag 消息解析失败，忽略: json={}, error={}", json, e.getMessage());
|
||||
return;
|
||||
}
|
||||
DemoMessageResult result = DemoMessageResult.from(message, consumer);
|
||||
store.add(result);
|
||||
log.info("Tag 消费成功 [{}]: id={}, tag={}, content={}",
|
||||
consumer, message.getId(), message.getTag(), message.getContent());
|
||||
}
|
||||
}
|
||||
@@ -26,12 +26,9 @@ public class DemoArtemisMessage {
|
||||
/// 消息内容
|
||||
private String content;
|
||||
|
||||
/// 消息场景（QUEUE / TOPIC / DELAY / TAG）
|
||||
/// 消息场景（QUEUE / TOPIC / DELAY）
|
||||
private String scene;
|
||||
|
||||
/// 消息标签（演示 Tag 过滤时使用，其它场景可为空）
|
||||
private String tag;
|
||||
|
||||
/// 发送时间（UTC）
|
||||
private OffsetDateTime sendTime;
|
||||
}
|
||||
|
||||
@@ -17,9 +17,9 @@ import lombok.experimental.Accessors;
|
||||
@Schema(title = "Artemis 演示消息发送参数")
|
||||
public class SendDemoMessageParam {
|
||||
|
||||
/// 消息场景：QUEUE / TOPIC / DELAY / TAG
|
||||
/// 消息场景：QUEUE / TOPIC / DELAY
|
||||
@NotNull(message = "{validation.field.scene.notNull}")
|
||||
@Schema(description = "消息场景: QUEUE/TOPIC/DELAY/TAG")
|
||||
@Schema(description = "消息场景: QUEUE/TOPIC/DELAY")
|
||||
private SendScene scene;
|
||||
|
||||
/// 消息内容
|
||||
@@ -27,10 +27,6 @@ public class SendDemoMessageParam {
|
||||
@Schema(description = "消息内容")
|
||||
private String content;
|
||||
|
||||
/// 消息标签（仅 TAG 场景必填，取值 important / normal）
|
||||
@Schema(description = "消息标签: important / normal（仅 TAG 场景使用）")
|
||||
private String tag;
|
||||
|
||||
/// 延时秒数（仅 DELAY 场景必填，范围 1-300）
|
||||
@Min(value = 1, message = "{validation.field.delaySeconds.min}")
|
||||
@Max(value = 300, message = "{validation.field.delaySeconds.max}")
|
||||
@@ -48,9 +44,6 @@ public class SendDemoMessageParam {
|
||||
TOPIC,
|
||||
|
||||
/// 延时消息
|
||||
DELAY,
|
||||
|
||||
/// Tag 过滤
|
||||
TAG
|
||||
DELAY
|
||||
}
|
||||
}
|
||||
|
||||
@@ -20,14 +20,10 @@ public class DemoMessageResult {
|
||||
@Schema(description = "业务消息ID")
|
||||
private String id;
|
||||
|
||||
/// 消息场景（QUEUE / TOPIC / DELAY / TAG）
|
||||
/// 消息场景（QUEUE / TOPIC / DELAY）
|
||||
@Schema(description = "消息场景")
|
||||
private String scene;
|
||||
|
||||
/// 消息标签
|
||||
@Schema(description = "消息标签")
|
||||
private String tag;
|
||||
|
||||
/// 消息内容
|
||||
@Schema(description = "消息内容")
|
||||
private String content;
|
||||
@@ -54,7 +50,6 @@ public class DemoMessageResult {
|
||||
DemoMessageResult result = new DemoMessageResult()
|
||||
.setId(message.getId())
|
||||
.setScene(message.getScene())
|
||||
.setTag(message.getTag())
|
||||
.setContent(message.getContent())
|
||||
.setSendTime(message.getSendTime())
|
||||
.setConsumeTime(now)
|
||||
|
||||
@@ -38,7 +38,7 @@ public class CacheInvalidationPublisher {
|
||||
try {
|
||||
// 缓存失效是广播语义，必须走 sendTopic（broker 端 cache-invalidation-topic 为 multicast 路由）
|
||||
artemisTemplateService.sendTopic(
|
||||
CacheTopicConstants.TOPIC, CacheTopicConstants.TAG_EVICT, JacksonUtil.toJson(message, false));
|
||||
CacheTopicConstants.TOPIC, JacksonUtil.toJson(message, false));
|
||||
log.debug("发布缓存失效消息成功: cacheName={}, key={}", cacheName, key);
|
||||
} catch (Exception e) {
|
||||
log.error("发布缓存失效消息失败: cacheName={}, key={}, error={}", cacheName, key, e.getMessage(), e);
|
||||
@@ -56,7 +56,7 @@ public class CacheInvalidationPublisher {
|
||||
try {
|
||||
// 缓存清空也是广播语义，同样使用 sendTopic
|
||||
artemisTemplateService.sendTopic(
|
||||
CacheTopicConstants.TOPIC, CacheTopicConstants.TAG_CLEAR, JacksonUtil.toJson(message, false));
|
||||
CacheTopicConstants.TOPIC, JacksonUtil.toJson(message, false));
|
||||
log.debug("发布缓存清空消息成功: cacheName={}", cacheName);
|
||||
} catch (Exception e) {
|
||||
log.error("发布缓存清空消息失败: cacheName={}, error={}", cacheName, e.getMessage(), e);
|
||||
|
||||
@@ -8,12 +8,6 @@ public interface CacheTopicConstants {
|
||||
/// 缓存失效通知 Topic
|
||||
String TOPIC = "cache-invalidation-topic";
|
||||
|
||||
/// 删除单个缓存键 Tag
|
||||
String TAG_EVICT = "evict";
|
||||
|
||||
/// 清空缓存 Tag
|
||||
String TAG_CLEAR = "clear";
|
||||
|
||||
/// 消费者组
|
||||
String CONSUMER_GROUP = "cache-invalidation-group";
|
||||
}
|
||||
|
||||
@@ -3,7 +3,6 @@ package org.dromara.daxpay.platform.common.artemis.service;
|
||||
import org.dromara.daxpay.platform.common.artemis.ArtemisCommonAutoConfiguration;
|
||||
import org.dromara.daxpay.platform.common.artemis.exception.ArtemisException;
|
||||
import cn.hutool.core.lang.UUID;
|
||||
import cn.hutool.core.util.StrUtil;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Qualifier;
|
||||
import org.springframework.jms.core.JmsTemplate;
|
||||
@@ -26,7 +25,6 @@ import java.util.Map;
|
||||
/// - `sendTopic*` 系列方法：发布订阅 Topic（multicast）
|
||||
/// - 延时消息使用 `JmsClient.withDeliveryDelay()`，对应 JMS 2.0 `MessageProducer.setDeliveryDelay()`
|
||||
/// - 业务幂等键通过 `KEYS` 属性传递（沿用 RocketMQ 时期的命名约定）
|
||||
/// - tag 作为消息属性保留，消费端可用 JMS selector `tag IN ('x','y')` 过滤
|
||||
///
|
||||
/// 为什么 Queue / Topic 必须分开方法：
|
||||
/// - `pubSubDomain` 是 JmsTemplate 实例级配置，决定 destination 解析方式
|
||||
@@ -38,7 +36,6 @@ import java.util.Map;
|
||||
/// - 发送端不参与对象转换，回落到 Spring 默认 `SimpleMessageConverter`：
|
||||
/// `String` payload 直接写入 `TextMessage`
|
||||
/// - 消费端 `@JmsListener` 方法签名统一 `onMessage(String json)`，自行反序列化
|
||||
/// - 这样消息保持自描述的 JSON 契约，类改名、跨服务消费、多语言接入都安全
|
||||
///
|
||||
/// @see JmsClient Spring Framework 7 fluent JMS 客户端
|
||||
@Slf4j
|
||||
@@ -48,9 +45,6 @@ public class ArtemisTemplateService {
|
||||
/// 业务幂等键属性名（沿用原 RocketMQ 命名，便于追踪）
|
||||
public static final String HEADER_KEYS = "KEYS";
|
||||
|
||||
/// 消息标签属性名，用于消费端 selector 过滤
|
||||
public static final String HEADER_TAG = "tag";
|
||||
|
||||
/// Queue 模板（pubSubDomain=false），由 Spring Boot 自动装配，Bean 名 `jmsTemplate`
|
||||
private final JmsTemplate queueJmsTemplate;
|
||||
|
||||
@@ -68,16 +62,15 @@ public class ArtemisTemplateService {
|
||||
/// 同步发送消息（点对点 Queue）
|
||||
///
|
||||
/// @param address 目标地址（对应 Artemis address，kebab-case 命名）
|
||||
/// @param tag 消息标签，用于消息过滤，为空时不设置
|
||||
/// @param body JSON 字符串消息体，由调用方自行序列化
|
||||
public void send(String address, String tag, String body) {
|
||||
Map<String, Object> headers = buildHeaders(tag);
|
||||
public void send(String address, String body) {
|
||||
Map<String, Object> headers = buildHeaders();
|
||||
try {
|
||||
jmsClient(queueJmsTemplate).destination(address).send(body, headers);
|
||||
log.debug("Queue 消息发送成功: address={}, tag={}, keys={}, body={}",
|
||||
address, tag, headers.get(HEADER_KEYS), body);
|
||||
log.debug("Queue 消息发送成功: address={}, keys={}, body={}",
|
||||
address, headers.get(HEADER_KEYS), body);
|
||||
} catch (MessagingException e) {
|
||||
log.error("Queue 消息发送异常: address={}, tag={}, error={}", address, tag, e.getMessage(), e);
|
||||
log.error("Queue 消息发送异常: address={}, error={}", address, e.getMessage(), e);
|
||||
throw new ArtemisException("error.artemis.sendFailed", e.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -88,21 +81,20 @@ public class ArtemisTemplateService {
|
||||
/// 底层等价于 `MessageProducer.setDeliveryDelay(delayMillis)`。
|
||||
///
|
||||
/// @param address 目标地址
|
||||
/// @param tag 消息标签
|
||||
/// @param body JSON 字符串消息体，由调用方自行序列化
|
||||
/// @param delaySeconds 延时秒数
|
||||
public void sendDelay(String address, String tag, String body, int delaySeconds) {
|
||||
Map<String, Object> headers = buildHeaders(tag);
|
||||
public void sendDelay(String address, String body, int delaySeconds) {
|
||||
Map<String, Object> headers = buildHeaders();
|
||||
long delayMillis = delaySeconds * 1000L;
|
||||
try {
|
||||
jmsClient(queueJmsTemplate).destination(address)
|
||||
.withDeliveryDelay(delayMillis)
|
||||
.send(body, headers);
|
||||
log.info("Queue 延时消息发送成功: address={}, tag={}, keys={}, delaySeconds={}, body={}",
|
||||
address, tag, headers.get(HEADER_KEYS), delaySeconds, body);
|
||||
log.info("Queue 延时消息发送成功: address={}, keys={}, delaySeconds={}, body={}",
|
||||
address, headers.get(HEADER_KEYS), delaySeconds, body);
|
||||
} catch (MessagingException e) {
|
||||
log.error("Queue 延时消息发送异常: address={}, tag={}, delaySeconds={}, error={}",
|
||||
address, tag, delaySeconds, e.getMessage(), e);
|
||||
log.error("Queue 延时消息发送异常: address={}, delaySeconds={}, error={}",
|
||||
address, delaySeconds, e.getMessage(), e);
|
||||
throw new ArtemisException("error.artemis.delaySendFailed", e.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -113,11 +105,10 @@ public class ArtemisTemplateService {
|
||||
/// 内部换算为相对当前时间的延时毫秒数。
|
||||
///
|
||||
/// @param address 目标地址
|
||||
/// @param tag 消息标签
|
||||
/// @param body JSON 字符串消息体，由调用方自行序列化
|
||||
/// @param deliveryTime 投递时刻（统一使用 OffsetDateTime，避免时区问题）
|
||||
public void sendDelayAt(String address, String tag, String body, OffsetDateTime deliveryTime) {
|
||||
Map<String, Object> headers = buildHeaders(tag);
|
||||
public void sendDelayAt(String address, String body, OffsetDateTime deliveryTime) {
|
||||
Map<String, Object> headers = buildHeaders();
|
||||
long deliveryTimestamp = deliveryTime.toInstant().toEpochMilli();
|
||||
long delayMillis = deliveryTimestamp - System.currentTimeMillis();
|
||||
if (delayMillis < 0) {
|
||||
@@ -128,11 +119,11 @@ public class ArtemisTemplateService {
|
||||
jmsClient(queueJmsTemplate).destination(address)
|
||||
.withDeliveryDelay(delayMillis)
|
||||
.send(body, headers);
|
||||
log.info("Queue 定时消息发送成功: address={}, tag={}, keys={}, deliveryTime={}, delayMillis={}",
|
||||
address, tag, headers.get(HEADER_KEYS), deliveryTime, delayMillis);
|
||||
log.info("Queue 定时消息发送成功: address={}, keys={}, deliveryTime={}, delayMillis={}",
|
||||
address, headers.get(HEADER_KEYS), deliveryTime, delayMillis);
|
||||
} catch (MessagingException e) {
|
||||
log.error("Queue 定时消息发送异常: address={}, tag={}, deliveryTime={}, error={}",
|
||||
address, tag, deliveryTime, e.getMessage(), e);
|
||||
log.error("Queue 定时消息发送异常: address={}, deliveryTime={}, error={}",
|
||||
address, deliveryTime, e.getMessage(), e);
|
||||
throw new ArtemisException("error.artemis.delaySendFailed", e.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -145,16 +136,15 @@ public class ArtemisTemplateService {
|
||||
/// `Destination ... does not support MULTICAST routing` 异常。
|
||||
///
|
||||
/// @param address 目标 Topic 地址
|
||||
/// @param tag 消息标签
|
||||
/// @param body JSON 字符串消息体，由调用方自行序列化
|
||||
public void sendTopic(String address, String tag, String body) {
|
||||
Map<String, Object> headers = buildHeaders(tag);
|
||||
public void sendTopic(String address, String body) {
|
||||
Map<String, Object> headers = buildHeaders();
|
||||
try {
|
||||
jmsClient(topicJmsTemplate).destination(address).send(body, headers);
|
||||
log.debug("Topic 消息发送成功: address={}, tag={}, keys={}, body={}",
|
||||
address, tag, headers.get(HEADER_KEYS), body);
|
||||
log.debug("Topic 消息发送成功: address={}, keys={}, body={}",
|
||||
address, headers.get(HEADER_KEYS), body);
|
||||
} catch (MessagingException e) {
|
||||
log.error("Topic 消息发送异常: address={}, tag={}, error={}", address, tag, e.getMessage(), e);
|
||||
log.error("Topic 消息发送异常: address={}, error={}", address, e.getMessage(), e);
|
||||
throw new ArtemisException("error.artemis.sendFailed", e.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -162,21 +152,20 @@ public class ArtemisTemplateService {
|
||||
/// 发送 Topic 延时消息
|
||||
///
|
||||
/// @param address 目标 Topic 地址
|
||||
/// @param tag 消息标签
|
||||
/// @param body JSON 字符串消息体，由调用方自行序列化
|
||||
/// @param delaySeconds 延时秒数
|
||||
public void sendTopicDelay(String address, String tag, String body, int delaySeconds) {
|
||||
Map<String, Object> headers = buildHeaders(tag);
|
||||
public void sendTopicDelay(String address, String body, int delaySeconds) {
|
||||
Map<String, Object> headers = buildHeaders();
|
||||
long delayMillis = delaySeconds * 1000L;
|
||||
try {
|
||||
jmsClient(topicJmsTemplate).destination(address)
|
||||
.withDeliveryDelay(delayMillis)
|
||||
.send(body, headers);
|
||||
log.info("Topic 延时消息发送成功: address={}, tag={}, keys={}, delaySeconds={}, body={}",
|
||||
address, tag, headers.get(HEADER_KEYS), delaySeconds, body);
|
||||
log.info("Topic 延时消息发送成功: address={}, keys={}, delaySeconds={}, body={}",
|
||||
address, headers.get(HEADER_KEYS), delaySeconds, body);
|
||||
} catch (MessagingException e) {
|
||||
log.error("Topic 延时消息发送异常: address={}, tag={}, delaySeconds={}, error={}",
|
||||
address, tag, delaySeconds, e.getMessage(), e);
|
||||
log.error("Topic 延时消息发送异常: address={}, delaySeconds={}, error={}",
|
||||
address, delaySeconds, e.getMessage(), e);
|
||||
throw new ArtemisException("error.artemis.delaySendFailed", e.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -184,11 +173,10 @@ public class ArtemisTemplateService {
|
||||
/// 发送 Topic 定时投递消息
|
||||
///
|
||||
/// @param address 目标 Topic 地址
|
||||
/// @param tag 消息标签
|
||||
/// @param body JSON 字符串消息体，由调用方自行序列化
|
||||
/// @param deliveryTime 投递时刻
|
||||
public void sendTopicDelayAt(String address, String tag, String body, OffsetDateTime deliveryTime) {
|
||||
Map<String, Object> headers = buildHeaders(tag);
|
||||
public void sendTopicDelayAt(String address, String body, OffsetDateTime deliveryTime) {
|
||||
Map<String, Object> headers = buildHeaders();
|
||||
long deliveryTimestamp = deliveryTime.toInstant().toEpochMilli();
|
||||
long delayMillis = deliveryTimestamp - System.currentTimeMillis();
|
||||
if (delayMillis < 0) {
|
||||
@@ -199,25 +187,22 @@ public class ArtemisTemplateService {
|
||||
jmsClient(topicJmsTemplate).destination(address)
|
||||
.withDeliveryDelay(delayMillis)
|
||||
.send(body, headers);
|
||||
log.info("Topic 定时消息发送成功: address={}, tag={}, keys={}, deliveryTime={}, delayMillis={}",
|
||||
address, tag, headers.get(HEADER_KEYS), deliveryTime, delayMillis);
|
||||
log.info("Topic 定时消息发送成功: address={}, keys={}, deliveryTime={}, delayMillis={}",
|
||||
address, headers.get(HEADER_KEYS), deliveryTime, delayMillis);
|
||||
} catch (MessagingException e) {
|
||||
log.error("Topic 定时消息发送异常: address={}, tag={}, deliveryTime={}, error={}",
|
||||
address, tag, deliveryTime, e.getMessage(), e);
|
||||
log.error("Topic 定时消息发送异常: address={}, deliveryTime={}, error={}",
|
||||
address, deliveryTime, e.getMessage(), e);
|
||||
throw new ArtemisException("error.artemis.delaySendFailed", e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
// ============================== 内部工具 ==============================
|
||||
|
||||
/// 构建消息头：业务幂等键 + 标签
|
||||
private Map<String, Object> buildHeaders(String tag) {
|
||||
Map<String, Object> headers = new HashMap<>(4);
|
||||
/// 构建消息头：业务幂等键
|
||||
private Map<String, Object> buildHeaders() {
|
||||
Map<String, Object> headers = new HashMap<>(2);
|
||||
// 随机 UUID 作为业务幂等键，去横线
|
||||
headers.put(HEADER_KEYS, UUID.randomUUID().toString(true));
|
||||
if (StrUtil.isNotEmpty(tag)) {
|
||||
headers.put(HEADER_TAG, tag);
|
||||
}
|
||||
return headers;
|
||||
}
|
||||
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
{
|
||||
"tagRequired": "Message tag is required for TAG scene",
|
||||
"delaySecondsRequired": "Delay seconds is required for DELAY scene"
|
||||
}
|
||||
|
||||
@@ -1,4 +1,3 @@
|
||||
{
|
||||
"tagRequired": "TAG 场景下消息标签不能为空",
|
||||
"delaySecondsRequired": "DELAY 场景下延时秒数不能为空"
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user