feat(integration): 添加 ZT-Message Kafka 生产者集成

- 新增 ZtMessage DTO 类用于消息体
- 实现 ZtMessageProducerService 生产者服务
- 添加示例演示如何发送消息
- 更新配置文件和文档以支持新功能
This commit is contained in:
2025-09-17 21:38:26 +08:00
parent dc2154c020
commit a888ed3fe2
6 changed files with 194 additions and 0 deletions

View File

@@ -0,0 +1,35 @@
package com.ycwl.basic.integration.message.dto;
import com.fasterxml.jackson.annotation.JsonInclude;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.HashMap;
import java.util.Map;
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@JsonInclude(JsonInclude.Include.NON_NULL)
public class ZtMessage {
private String channelId; // required
private String title; // required
private String content; // required
private String target; // required
private Map<String, Object> extra; // optional
private String sendReason; // optional
private String sendBiz; // optional
public static ZtMessage of(String channelId, String title, String content, String target) {
return ZtMessage.builder()
.channelId(channelId)
.title(title)
.content(content)
.target(target)
.extra(new HashMap<>())
.build();
}
}

View File

@@ -0,0 +1,33 @@
package com.ycwl.basic.integration.message.example;
import com.ycwl.basic.integration.message.dto.ZtMessage;
import com.ycwl.basic.integration.message.service.ZtMessageProducerService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Component;
import java.util.HashMap;
@Slf4j
@Component
@RequiredArgsConstructor
@ConditionalOnProperty(name = "kafka.example.zt-message.enabled", havingValue = "true", matchIfMissing = false)
public class ZtMessageProducerExample {
private final ZtMessageProducerService producer;
public void demo() {
log.info("=== ZT-Message Producer Example ===");
ZtMessage msg = ZtMessage.of("dummy", "标题", "内容", "user-001");
var extra = new HashMap<String, Object>();
extra.put("k", "v");
msg.setExtra(extra);
msg.setSendReason("REGISTER");
msg.setSendBiz("USER");
producer.send(msg);
log.info("ZT-Message produced");
}
}

View File

@@ -0,0 +1,64 @@
package com.ycwl.basic.integration.message.service;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.ycwl.basic.integration.kafka.config.KafkaIntegrationProperties;
import com.ycwl.basic.integration.message.dto.ZtMessage;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import java.util.Map;
@Slf4j
@Service
@RequiredArgsConstructor
@ConditionalOnProperty(name = "kafka.enabled", havingValue = "true")
public class ZtMessageProducerService {
public static final String DEFAULT_TOPIC = "zt-message";
private final KafkaTemplate<String, String> kafkaTemplate;
private final ObjectMapper objectMapper;
private final KafkaIntegrationProperties kafkaProps;
public void send(ZtMessage msg) {
validate(msg);
String topic = kafkaProps != null && StringUtils.isNotBlank(kafkaProps.getZtMessageTopic())
? kafkaProps.getZtMessageTopic()
: DEFAULT_TOPIC;
String key = msg.getChannelId();
String payload = toJson(msg);
log.info("[ZT-MESSAGE] producing to topic={}, key={}, title={}", topic, key, msg.getTitle());
kafkaTemplate.send(topic, key, payload).whenComplete((metadata, ex) -> {
if (ex != null) {
log.error("[ZT-MESSAGE] produce failed: {}", ex.getMessage(), ex);
} else if (metadata != null) {
log.info("[ZT-MESSAGE] produced: partition={}, offset={}", metadata.getRecordMetadata().partition(), metadata.getRecordMetadata().offset());
}
});
}
private void validate(ZtMessage msg) {
if (msg == null) throw new IllegalArgumentException("message is null");
if (StringUtils.isBlank(msg.getChannelId())) throw new IllegalArgumentException("channelId is required");
if (StringUtils.isBlank(msg.getTitle())) throw new IllegalArgumentException("title is required");
if (StringUtils.isBlank(msg.getContent())) throw new IllegalArgumentException("content is required");
if (StringUtils.isBlank(msg.getTarget())) throw new IllegalArgumentException("target is required");
if (msg.getExtra() != null && !(msg.getExtra() instanceof Map)) {
throw new IllegalArgumentException("extra must be a Map");
}
}
private String toJson(ZtMessage msg) {
try {
return objectMapper.writeValueAsString(msg);
} catch (JsonProcessingException e) {
throw new IllegalArgumentException("failed to serialize message", e);
}
}
}