推送订阅功能

This commit is contained in:
chenxing
2026-08-14 09:05:34 +08:00
parent e55c2cb209
commit 8ff6e7ac0c
14 changed files with 553 additions and 0 deletions
@@ -0,0 +1,102 @@
package com.yunhai.userbase.config;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import com.yunhai.userbase.model.push.PushTaskMessage;
import java.util.HashMap;
import java.util.Map;
/**
* 推送订阅 MQ 配置。<br><br>
* 拓扑:
* <pre>
* push.topic (topic, durable)
* └─ push.main (durable, concurrency=1, x-dead-letter-exchange=push.dlx) ← 串行消费,可观测
* push.dlx (fanout, durable)
* └─ push.dlq (durable) ← 死信归档(重试耗尽)
* </pre>
* 连接信息(host/port/user/pwd/vhost)由 Nacos {@code application-infra.yaml} 的 {@code spring.rabbitmq.*}
* 公共配置注入,本类不写死(也不改 bootstrap.yml / application.yml)。<br>
* 可靠性:消费端手动 ack,失败由 {@code PushConsumer} 应用层自管重试(共 3 次),耗尽进死信队列。
*
* @author chenx
* @date 2026/8/14
*/
@Configuration
public class PushMqConfig {
public static final String EXCHANGE_TOPIC = "push.topic";
public static final String EXCHANGE_DLX = "push.dlx";
public static final String QUEUE_MAIN = "push.main";
public static final String QUEUE_DLQ = "push.dlq";
/** 统一 JSON 消息转换器(固定反序列化为 PushTaskMessage,绕过 trusted-packages 版本差异)。 */
@Bean
public MessageConverter pushMessageConverter() {
Jackson2JsonMessageConverter converter = new Jackson2JsonMessageConverter();
converter.setClassMapper(new PushTaskClassMapper(PushTaskMessage.class));
return converter;
}
@Bean
public TopicExchange pushTopicExchange() {
return new TopicExchange(EXCHANGE_TOPIC, true, false);
}
@Bean
public FanoutExchange pushDlxExchange() {
return new FanoutExchange(EXCHANGE_DLX, true, false);
}
/** 主队列:单消费者串行 + 死信路由到 push.dlx。 */
@Bean
public Queue pushMainQueue() {
Map<String, Object> args = new HashMap<>();
args.put("x-dead-letter-exchange", EXCHANGE_DLX);
return new Queue(QUEUE_MAIN, true, false, false, args);
}
/** 死信队列:归档重试耗尽的任务。 */
@Bean
public Queue pushDlqQueue() {
return new Queue(QUEUE_DLQ, true, false, false);
}
@Bean
public Binding pushMainBinding() {
return BindingBuilder.bind(pushMainQueue()).to(pushTopicExchange()).with("#");
}
@Bean
public Binding pushDlqBinding() {
return BindingBuilder.bind(pushDlqQueue()).to(pushDlxExchange());
}
@Bean
public RabbitTemplate pushRabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate template = new RabbitTemplate(connectionFactory);
template.setMessageConverter(pushMessageConverter());
return template;
}
/** 消费者容器:手动 ack + 单并发 + prefetch=1,保证串行可观测。 */
@Bean(name = "pushListenerContainerFactory")
public SimpleRabbitListenerContainerFactory pushListenerContainerFactory(ConnectionFactory connectionFactory) {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setConnectionFactory(connectionFactory);
factory.setMessageConverter(pushMessageConverter());
factory.setAcknowledgeMode(org.springframework.amqp.core.AcknowledgeMode.MANUAL);
factory.setConcurrentConsumers(1);
factory.setMaxConcurrentConsumers(1);
factory.setPrefetchCount(1);
return factory;
}
}
@@ -0,0 +1,32 @@
package com.yunhai.userbase.config;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.support.converter.ClassMapper;
/**
* 推送消息固定反序列化为 {@link PushTaskMessage},绕过 Jackson2JsonMessageConverter 的
* trusted-packages 安全限制(不同 spring-amqp 版本该方法签名/默认值不一致,编译期不可靠)。<br>
* 生产者端序列化仍由 Jackson2JsonMessageConverter 完成(写入 __TypeId__),消费者端忽略 __TypeId__
* 直接按本映射器返回 PushTaskMessage,类型单一且可控。
*
* @author chenx
* @date 2026/8/14
*/
public class PushTaskClassMapper implements ClassMapper {
private final Class<?> payloadClass;
public PushTaskClassMapper(Class<?> payloadClass) {
this.payloadClass = payloadClass;
}
@Override
public Class<?> toClass(MessageProperties properties) {
return payloadClass;
}
@Override
public void fromClass(Class<?> clazz, MessageProperties properties) {
properties.setHeader("__TypeId__", clazz.getName());
}
}
@@ -0,0 +1,25 @@
package com.yunhai.userbase.event;
import com.yunhai.userbase.model.push.PushInterfaceType;
import lombok.Data;
import java.util.List;
/**
* 数据同步事件:组 / 用户变更后由 {@code SysGroupService}/{@code SysUserService} 在事务内 publish,
* {@code PushDispatcher} 用 {@code @TransactionalEventListener(AFTER_COMMIT)} 消费(事务提交后才推送,
* 推送失败不影响主业务)。
*
* @author chenx
* @date 2026/8/14
*/
@Data
public class DataSyncEvent {
/** 接口类型,如 group/AddGroup */
private PushInterfaceType type;
/** 组事件携带的受影响组 id 列表 */
private List<Integer> groupIds;
/** 用户事件携带的受影响用户 id 列表 */
private List<Integer> userIds;
}
@@ -0,0 +1,30 @@
package com.yunhai.userbase.model.application;
import lombok.Data;
/**
* 保存应用推送设置请求(应用列表「推送设置」按钮提交)。<br>
* 所有字段均为可选更新:传 null 表示不修改该项。
*
* @author chenx
* @date 2026/8/14
*/
@Data
public class PushSettingRequest {
/** 推送外部接口地址(不含固定后缀) */
private String pushBaseUrl;
/** 推送总开关:0关 1开 */
private Integer pushEnabled;
/** 启用的接口路径,逗号分隔 */
private String pushEventTypes;
/** 关联学校组 id,逗号分隔;空=全局 */
private String pushSchoolIds;
/** 推送角色 key,逗号分隔;空=全部角色 */
private String pushRoleIds;
/**
* 推送签名密钥;传 null 表示不修改(保留现有值),
* 传空串 "" 表示清空(后续复用 appSecret)。
*/
private String pushSecret;
}
@@ -0,0 +1,24 @@
package com.yunhai.userbase.model.application;
import lombok.Data;
/**
* 应用推送设置回显 VO(GetPushSetting 返回,仅含推送相关字段,不含 appSecret 等无关大字段)。
*
* @author chenx
* @date 2026/8/14
*/
@Data
public class PushSettingVo {
private Integer id;
private Integer appKey;
private String name;
private String pushBaseUrl;
private Integer pushEnabled;
private String pushEventTypes;
private String pushSchoolIds;
private String pushRoleIds;
/** 推送签名密钥(可展示供复制;空=复用 appSecret) */
private String pushSecret;
}
@@ -0,0 +1,49 @@
package com.yunhai.userbase.model.push;
import lombok.AllArgsConstructor;
import lombok.Getter;
/**
* 文档约定的 8 个订阅接口:path(固定后缀,拼在订阅方 base URL 后)↔ 友好名 label。
*
* @author chenx
* @date 2026/8/14
*/
@Getter
@AllArgsConstructor
public enum PushInterfaceType {
SYNC_GROUP("group/SyncGroup", "组全量同步"),
SYNC_USER("user/SyncUser", "用户全量同步"),
ADD_GROUP("group/AddGroup", "添加组"),
UPDATE_GROUP("group/UpdateGroup", "更新组"),
DELETE_GROUP("group/DeleteGroup", "删除组"),
ADD_USER("user/AddUser", "添加用户"),
UPDATE_USER("user/UpdateUser", "更新用户"),
DELETE_USER("user/DeleteUser", "删除用户");
private final String path;
private final String label;
public static PushInterfaceType getByPath(String path) {
if (path == null) {
return null;
}
for (PushInterfaceType type : values()) {
if (type.path.equals(path)) {
return type;
}
}
return null;
}
/** 是否为组相关接口。 */
public boolean isGroup() {
return path.startsWith("group/");
}
/** 是否为用户相关接口。 */
public boolean isUser() {
return path.startsWith("user/");
}
}
@@ -0,0 +1,74 @@
package com.yunhai.userbase.model.push;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.hibernate.annotations.CreationTimestamp;
import javax.persistence.*;
import java.time.LocalDateTime;
/**
* 推送订阅记录实体(订阅方维度,落库用于日志 / 重试 / 排查)。<br>
* 与 DDL(push_subscription_ddl.sql §2)的 push_record 表一一对应。
*
* @author chenx
* @date 2026/8/14
*/
@Data
@NoArgsConstructor
@Entity
@Table(name = "push_record")
public class PushRecordEntity {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(name = "subscriber_app_key", nullable = false)
private Integer subscriberAppKey;
@Column(name = "app_name")
private String appName;
@Column(name = "interface_path", nullable = false)
private String interfacePath;
@Column(name = "interface_label")
private String interfaceLabel;
@Column(name = "request_url", nullable = false)
private String requestUrl;
@Column(name = "payload", columnDefinition = "MEDIUMTEXT")
private String payload;
@Column(name = "sign_t")
private String signT;
@Column(name = "http_status")
private Integer httpStatus;
@Column(name = "resp_code")
private Integer respCode;
@Column(name = "resp_msg")
private String respMsg;
/** 是否成功:0失败 1成功 */
@Column(name = "success", nullable = false)
private Integer success = 0;
@Column(name = "attempt_no", nullable = false)
private Integer attemptNo = 1;
@Column(name = "cost_ms")
private Integer costMs;
/** 触发类型:manual / auto / retry */
@Column(name = "trigger_type")
private String triggerType;
@CreationTimestamp
@Column(name = "created_at", nullable = false, updatable = false)
private LocalDateTime createdAt;
}
@@ -0,0 +1,24 @@
package com.yunhai.userbase.model.push;
import lombok.Data;
/**
* 推送记录查询条件(前端日志页搜索用)。
*
* @author chenx
* @date 2026/8/14
*/
@Data
public class PushRecordQuery {
/** 订阅方 appKey */
private Integer appKey;
/** 接口路径,如 group/AddGroup */
private String interfacePath;
/** 是否成功:0失败 1成功 */
private Integer success;
/** 起始时间 yyyy-MM-dd(含当天 00:00:00) */
private String startTime;
/** 结束时间 yyyy-MM-dd(含当天 23:59:59) */
private String endTime;
}
@@ -0,0 +1,38 @@
package com.yunhai.userbase.model.push;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* MQ 推送任务消息。<br>
* 由 {@code PushDispatcher}(生产方)投递,{@code PushConsumer}(消费方)取出后 POST 给订阅方。<br>
* 签名(t + sign)在投递时由基础平台计算一次;自动重试保持原 t/sign,手动重试重新生成。
*
* @author chenx
* @date 2026/8/14
*/
@Data
@NoArgsConstructor
@AllArgsConstructor
public class PushTaskMessage implements Serializable {
/** 订阅方 appKey */
private Integer appKey;
/** 接口路径,如 group/AddGroup */
private String interfacePath;
/** 接口友好名 */
private String interfaceLabel;
/** 业务数据原始 JSON 字符串(签名与发送均基于该原始字符串字节) */
private String body;
/** 签名时间戳(字符串) */
private String t;
/** HMAC-SHA256(业务JSON + t, secret) hex 小写 */
private String sign;
/** 第几次尝试:1=首次 */
private int attemptNo;
/** 触发类型:manual / auto / retry */
private String triggerType;
}
@@ -0,0 +1,22 @@
package com.yunhai.userbase.model.push;
import lombok.Data;
import java.util.List;
/**
* 测试推送请求(测试推送页调用)。
*
* @author chenx
* @date 2026/8/14
*/
@Data
public class PushTestRequest {
/** 订阅方应用 id */
private Integer appId;
/** 接口路径,如 group/AddGroup */
private String interfacePath;
/** 目标 id 列表(可选):不传则按接口类型取全量/全部 */
private List<Integer> targetIds;
}
@@ -0,0 +1,32 @@
package com.yunhai.userbase.model.push.dto;
import lombok.Data;
import java.util.List;
/**
* 文档「组」契约 DTO(group/SyncGroup、AddGroup、UpdateGroup 共用)。<br>
* 字段严格对齐《弘育基础平台数据同步接口 2》:
* <ul>
* <li>SyncGroup:children 递归树(可空)</li>
* <li>AddGroup:pid 必填(上级组 id,0 为顶级)</li>
* <li>UpdateGroup:仅 id 必填,其余可选</li>
* </ul>
*
* @author chenx
* @date 2026/8/14
*/
@Data
public class PushGroupDTO {
private Integer id;
private String name;
private String code;
/** 组类别 1-8,与 GroupTypeEnum 1:1 零映射 */
private Integer type;
/** 上级组 id(AddGroup 必填) */
private Integer pid;
private Integer no;
/** 子组列表(SyncGroup 用) */
private List<PushGroupDTO> children;
}
@@ -0,0 +1,53 @@
package com.yunhai.userbase.model.push.dto;
import lombok.Data;
import java.util.List;
/**
* 文档「用户」契约 DTO(user/SyncUser、AddUser、UpdateUser 共用)。<br>
* 字段严格对齐《弘育基础平台数据同步接口 2》:
* <ul>
* <li>status:0正常 / 2禁用(由 userbase StatusEnum 映射)</li>
* <li>groups:用户所属组,含 groupId / no / schoolId(schoolId = 组所属学校范围 id)</li>
* <li>roleList:用户身份,含 roleId / schId(schId = 用户主组所属学校范围 id)</li>
* </ul>
*
* @author chenx
* @date 2026/8/14
*/
@Data
public class PushUserDTO {
private Integer id;
private String account;
private String name;
private String idcard;
private String phone;
private String email;
/** 性别 0女 1男 */
private Integer gender;
private String birthday;
/** 用户状态 0正常 2禁用 */
private Integer status;
private String profilePicture;
private List<PushUserGroupRef> groups;
private List<PushUserRoleRef> roleList;
/** 用户所属组引用。 */
@Data
public static class PushUserGroupRef {
private Integer groupId;
private Integer no;
/** 组所属学校范围 id(沿 parentId 向上最近的 SCHOOL_GROUP) */
private Integer schoolId;
}
/** 用户身份引用。 */
@Data
public static class PushUserRoleRef {
private Integer roleId;
/** 所属学校 id(userbase 无角色-学校关联,统一取用户主组所属学校) */
private Integer schId;
}
}
@@ -0,0 +1,16 @@
package com.yunhai.userbase.repository;
import com.yunhai.userbase.model.push.PushRecordEntity;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
import org.springframework.stereotype.Repository;
/**
* 推送记录仓储。
*
* @author chenx
* @date 2026/8/14
*/
@Repository
public interface PushRecordRepository extends JpaRepository<PushRecordEntity, Long>, JpaSpecificationExecutor<PushRecordEntity> {
}
@@ -0,0 +1,32 @@
package com.yunhai.userbase.task;
import com.yunhai.userbase.service.push.PushRecordService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
/**
* 推送日志定时清理任务:每日凌晨 3 点清理 1 年前的推送记录。<br>
* 与 {@code OperationLogCleanupTask} 同范式(@EnableScheduling 已在 UserBaseApplication 开启)。
*
* @author chenx
* @date 2026/8/14
*/
@Slf4j
@Component
public class PushRecordCleanupTask {
@Autowired
private PushRecordService pushRecordService;
@Scheduled(cron = "0 0 3 * * ?")
public void run() {
try {
long deleted = pushRecordService.cleanupExpired();
log.info("推送日志定时清理完成, 删除{}条", deleted);
} catch (Exception e) {
log.error("推送日志定时清理失败", e);
}
}
}