feat(mq): 重构用户同步为MQ消息驱动并移除HTTP同步服务

- 将用户同步消息交换机由 DirectExchange 改为 TopicExchange，实现多目标系统路由
- 去除核心系统中用户同步队列定义和消费者，转由各子系统独立消费
- RabbitMQSyncProducer 动态构建 routing key，支持按目标系统路由同步消息
- UserServiceImpl 用户新增和更新后发送同步消息到 websopy 的 MQ 交换机
- 废弃 UserSyncService 中的 HTTP 同步接口，改为通过 MQ 实现用户数据同步
- 删除 UserSyncConsumer，核心系统不再直接消费用户同步消息
- 增加日志输出，便于追踪用户同步消息发送情况
- 保留废弃代码兼容旧引用，方便后续平滑迁移和维护

src/main/java/com/gxwebsoft/common/mq/config/RabbitMQConfig.java

@@ -1,7 +1,13 @@
 package com.gxwebsoft.common.mq.config;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
-import org.springframework.amqp.core.*;
+import org.springframework.amqp.core.DirectExchange;
+import org.springframework.amqp.core.TopicExchange;
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.core.QueueBuilder;
+import org.springframework.amqp.core.Binding;
+import org.springframework.amqp.core.BindingBuilder;
+import org.springframework.amqp.core.AcknowledgeMode;
 import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
 import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
 import org.springframework.amqp.rabbit.connection.ConnectionFactory;
@@ -98,9 +104,15 @@ public class RabbitMQConfig {
 
     // ==================== 交换机 ====================
 
+    /**
+     * 用户同步 Topic Exchange
+     * 使用 Topic 类型，支持按 targetSystem 路由到不同队列
+     * routing key 格式: user.sync.{targetSystem}
+     * 各子系统可以绑定自己的队列来消费消息
+     */
     @Bean
-    public DirectExchange syncExchange() {
-        return new DirectExchange(SYNC_EXCHANGE, true, false);
+    public TopicExchange syncExchange() {
+        return new TopicExchange(SYNC_EXCHANGE, true, false);
     }
 
     @Bean
@@ -110,13 +122,12 @@ public class RabbitMQConfig {
 
     // ==================== 队列 ====================
 
-    @Bean
-    public Queue syncQueue() {
-        return QueueBuilder.durable(SYNC_QUEUE)
-                .withArgument("x-dead-letter-exchange", DLX_EXCHANGE)
-                .withArgument("x-dead-letter-routing-key", DLQ_ROUTING_KEY)
-                .build();
-    }
+    /**
+     * 注意：core 系统只负责发送消息，不消费消息
+     * 各子系统（websopy等）需要在自己的系统中配置消费者和队列
+     * 
+     * 如果 core 系统也需要消费某些消息，可以在这里添加对应的队列
+     */
 
     @Bean
     public Queue deadLetterQueue() {
@@ -125,13 +136,6 @@ public class RabbitMQConfig {
 
     // ==================== 绑定 ====================
 
-    @Bean
-    public Binding syncBinding() {
-        return BindingBuilder.bind(syncQueue())
-                .to(syncExchange())
-                .with(SYNC_ROUTING_KEY);
-    }
-
     @Bean
     public Binding dlqBinding() {
         return BindingBuilder.bind(deadLetterQueue())

src/main/java/com/gxwebsoft/common/mq/consumer/UserSyncConsumer.java

@@ -1,159 +0,0 @@
-package com.gxwebsoft.common.mq.consumer;
-
-import com.gxwebsoft.common.mq.config.RabbitMQConfig;
-import com.gxwebsoft.common.mq.message.SyncMessage;
-import com.gxwebsoft.common.system.service.UserSyncService;
-import com.rabbitmq.client.Channel;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.amqp.core.Message;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
-import org.springframework.stereotype.Component;
-
-import java.io.IOException;
-import java.util.Map;
-
-/**
- * 用户同步消息消费者
- * 
- * 负责监听用户同步消息，并将数据同步到目标系统（如websopy）
- */
-@Slf4j
-@Component
-@ConditionalOnProperty(name = "sync.mq.enabled", havingValue = "true", matchIfMissing = true)
-public class UserSyncConsumer {
-
-    @Autowired(required = false)
-    private UserSyncService userSyncService;
-
-    private static final int MAX_RETRY_COUNT = 3;
-
-    /**
-     * 监听用户同步消息
-     */
-    @RabbitListener(queues = RabbitMQConfig.SYNC_QUEUE)
-    public void handleMessage(SyncMessage message, Channel channel, Message amqpMessage) {
-        long deliveryTag = amqpMessage.getMessageProperties().getDeliveryTag();
-
-        try {
-            log.info("收到MQ消息: messageId={}, type={}, event={}, target={}",
-                    message.getMessageId(), message.getMessageType(),
-                    message.getEventType(), message.getTargetSystem());
-
-            // 检查是否启用了同步服务
-            if (userSyncService == null) {
-                log.warn("UserSyncService 未启用，跳过消息处理: messageId={}", message.getMessageId());
-                channel.basicAck(deliveryTag, false);
-                return;
-            }
-
-            // 处理消息
-            processMessage(message);
-
-            // 确认消息
-            channel.basicAck(deliveryTag, false);
-            log.info("消息处理成功: messageId={}", message.getMessageId());
-
-        } catch (Exception e) {
-            log.error("消息处理失败: messageId={}, error={}", message.getMessageId(), e.getMessage(), e);
-            handleFailure(channel, amqpMessage, deliveryTag, message, e);
-        }
-    }
-
-    /**
-     * 处理消息
-     */
-    private void processMessage(SyncMessage message) {
-        String messageType = message.getMessageType();
-        String eventType = message.getEventType();
-        String targetSystem = message.getTargetSystem();
-        Map<String, Object> data = message.getData();
-
-        // 判断消息类型并处理
-        if ("USER_SYNC".equals(messageType)) {
-            handleUserSync(targetSystem, eventType, data);
-        } else {
-            log.warn("未知的消息类型: messageType={}", messageType);
-        }
-    }
-
-    /**
-     * 处理用户同步
-     */
-    private void handleUserSync(String targetSystem, String eventType, Map<String, Object> data) {
-        switch (eventType) {
-            case "CREATE":
-            case "UPDATE":
-                userSyncService.syncUserToWebsopy(data);
-                log.info("用户同步到{}成功: event={}", targetSystem, eventType);
-                break;
-            case "DELETE":
-                userSyncService.deleteUserFromWebsopy(data);
-                log.info("用户从{}删除成功: event={}", targetSystem, eventType);
-                break;
-            default:
-                log.warn("未知的用户事件类型: eventType={}", eventType);
-        }
-    }
-
-    /**
-     * 处理失败消息
-     */
-    private void handleFailure(Channel channel, Message amqpMessage, long deliveryTag,
-                               SyncMessage message, Exception e) {
-        Integer retryCount = message.getRetryCount();
-        if (retryCount == null) {
-            retryCount = 0;
-        }
-
-        if (retryCount < MAX_RETRY_COUNT) {
-            // 重试：拒绝消息并重新入队
-            try {
-                log.warn("消息处理失败，准备重试: messageId={}, retryCount={}/{}",
-                        message.getMessageId(), retryCount + 1, MAX_RETRY_COUNT);
-                // 增加重试次数
-                message.setRetryCount(retryCount + 1);
-                // 拒绝消息（requeue=true 会重新入队）
-                channel.basicNack(deliveryTag, false, true);
-            } catch (IOException ioException) {
-                log.error("消息拒绝失败: messageId={}", message.getMessageId(), ioException);
-            }
-        } else {
-            // 超过重试次数，发送到死信队列
-            try {
-                log.error("消息处理失败次数超限，发送到死信队列: messageId={}, retryCount={}",
-                        message.getMessageId(), retryCount);
-                // 拒绝消息（requeue=false 进入死信队列）
-                channel.basicNack(deliveryTag, false, false);
-            } catch (IOException ioException) {
-                log.error("消息拒绝失败: messageId={}", message.getMessageId(), ioException);
-            }
-        }
-    }
-
-    /**
-     * 监听死信队列消息（可选，用于告警或人工处理）
-     */
-    @RabbitListener(queues = RabbitMQConfig.DLQ_QUEUE)
-    public void handleDeadLetter(SyncMessage message, Channel channel, Message amqpMessage) {
-        long deliveryTag = amqpMessage.getMessageProperties().getDeliveryTag();
-        try {
-            log.error("死信消息: messageId={}, type={}, event={}, target={}, retryCount={}",
-                    message.getMessageId(), message.getMessageType(),
-                    message.getEventType(), message.getTargetSystem(),
-                    message.getRetryCount());
-
-            // TODO: 可以在这里添加告警逻辑，如发送邮件、钉钉通知等
-
-            channel.basicAck(deliveryTag, false);
-        } catch (Exception e) {
-            log.error("处理死信消息失败: messageId={}", message.getMessageId(), e);
-            try {
-                channel.basicAck(deliveryTag, false);
-            } catch (IOException ioException) {
-                log.error("确认死信消息失败", ioException);
-            }
-        }
-    }
-}

src/main/java/com/gxwebsoft/common/mq/producer/impl/RabbitMQSyncProducer.java

@@ -66,9 +66,14 @@ public class RabbitMQSyncProducer implements SyncMessageProducer, RabbitTemplate
                 );
             }
 
+            // 使用 targetSystem 作为 routing key
+            // 格式: user.sync.{targetSystem}
+            // 各子系统绑定队列时使用 pattern: user.sync.{systemName}
+            String routingKey = buildRoutingKey(message.getTargetSystem());
+            
             rabbitTemplate.convertAndSend(
                     RabbitMQConfig.SYNC_EXCHANGE,
-                    RabbitMQConfig.SYNC_ROUTING_KEY,
+                    routingKey,
                     message,
                     correlationData
             );
@@ -81,6 +86,17 @@ public class RabbitMQSyncProducer implements SyncMessageProducer, RabbitTemplate
         }
     }
 
+    /**
+     * 构建 routing key
+     * 格式: user.sync.{targetSystem}
+     */
+    private String buildRoutingKey(String targetSystem) {
+        if (targetSystem == null || targetSystem.isEmpty()) {
+            return "user.sync.all";
+        }
+        return "user.sync." + targetSystem.toLowerCase();
+    }
+
     @Override
     public void sendUserSyncMessage(String targetSystem, String eventType, Object userData) {
         try {

src/main/java/com/gxwebsoft/common/system/service/UserSyncService.java

@@ -1,302 +1,42 @@
 package com.gxwebsoft.common.system.service;
 
-import com.gxwebsoft.common.core.config.ConfigProperties;
-import com.gxwebsoft.common.core.utils.HttpUtils;
-import com.gxwebsoft.common.system.entity.User;
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.http.HttpResponse;
-import org.apache.http.util.EntityUtils;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.Resource;
-import java.nio.charset.StandardCharsets;
-import java.time.LocalDateTime;
-import java.time.format.DateTimeFormatter;
-import java.util.HashMap;
-import java.util.Map;
-
 /**
- * 用户同步服务（同步到 websopy）
+ * 用户同步服务
+ * 
+ * 注意：此服务已废弃，用户同步现在通过 MQ 实现。
+ * 各子系统（websopy等）需要在自己的系统中消费 MQ 消息进行同步。
+ * 
+ * 保留此类是为了兼容可能存在的旧代码引用，所有方法已改为空实现。
  *
  * @author WebSoft
  * @since 2026-04-04
+ * @deprecated 请使用 MQ 消息进行用户同步
  */
 @Slf4j
 @Service
+@Deprecated
 public class UserSyncService {
-    
-    /**
-     * 转义JSON字符串中的特殊字符
-     */
-    private String escapeJson(String str) {
-        if (str == null) {
-            return "";
-        }
-        return str.replace("\\", "\\\\")
-                  .replace("\"", "\\\"")
-                  .replace("\b", "\\b")
-                  .replace("\f", "\\f")
-                  .replace("\n", "\\n")
-                  .replace("\r", "\\r")
-                  .replace("\t", "\\t");
-    }
-
-    @Resource
-    private ConfigProperties configProperties;
-
-    /**
-     * 获取 websopy 基础 URL
-     */
-    private String getWebsopyBaseUrl() {
-        return configProperties.getWebsopyUrl();
-    }
 
     /**
      * 同步单个用户到 websopy
-     *
-     * @param user 用户信息
+     * 
+     * @deprecated 已废弃，用户同步现在通过 MQ 自动触发
      */
-    public void syncUserToWebsopy(User user) {
-        String websopyBaseUrl = getWebsopyBaseUrl();
-        if (websopyBaseUrl == null || websopyBaseUrl.isEmpty()) {
-            log.warn("websopyUrl 未配置，跳过用户同步: userId={}", user.getUserId());
-            return;
-        }
-
-        if (user == null || user.getUserId() == null) {
-            log.warn("用户信息为空，跳过同步");
-            return;
-        }
-
-        try {
-            // 构建请求体 - 使用手动JSON构建确保字段正确
-            Integer tenantIdValue = user.getTenantId() != null ? user.getTenantId() : 0;
-            
-            // 构建JSON字符串，确保tenant_id字段存在且不为null
-            StringBuilder jsonBuilder = new StringBuilder();
-            jsonBuilder.append("{");
-            jsonBuilder.append("\"userId\":").append(user.getUserId()).append(",");
-            jsonBuilder.append("\"username\":\"").append(escapeJson(user.getUsername())).append("\",");
-            jsonBuilder.append("\"nickname\":\"").append(escapeJson(user.getNickname())).append("\",");
-            jsonBuilder.append("\"avatar\":\"").append(escapeJson(user.getAvatar())).append("\",");
-            jsonBuilder.append("\"phone\":\"").append(escapeJson(user.getPhone())).append("\",");
-            jsonBuilder.append("\"status\":").append(user.getStatus()).append(",");
-            jsonBuilder.append("\"updateTime\":\"").append(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))).append("\",");
-            jsonBuilder.append("\"tenantId\":").append(tenantIdValue).append(",");  // 驼峰格式
-            jsonBuilder.append("\"tenant_id\":").append(tenantIdValue);             // 下划线格式
-            jsonBuilder.append("}");
-            
-            String url = websopyBaseUrl + "/api/app/user-sync/single";
-            String body = jsonBuilder.toString();
-
-            log.info("同步用户到 websopy: userId={}, username={}, nickname={}, phone={}, tenantId={}, url={}", 
-                    user.getUserId(), user.getUsername(), user.getNickname(), user.getPhone(), user.getTenantId(), url);
-            log.info("同步用户请求体JSON: {}", body);  // 改为info级别以便查看
-            // 额外日志：tenantId 值检查
-            log.debug("tenantId检查 - 原始值: {}, 转换后值: {}", 
-                    user.getTenantId(), tenantIdValue);
-
-            // 发送 HTTP POST 请求
-            Map<String, String> headers = new HashMap<>();
-            headers.put("Content-Type", "application/json");
-
-            HttpResponse response = HttpUtils.doPost(url, "", "POST", headers, null, body);
-            String responseBody = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
-
-            log.debug("websopy 响应: {}", responseBody);
-
-            // 解析响应
-            JSONObject result = JSON.parseObject(responseBody);
-            if (result != null && result.getIntValue("code") == 0) {
-                log.info("用户同步成功: userId={}", user.getUserId());
-            } else {
-                String message = result != null ? result.getString("message") : "未知错误";
-                log.error("用户同步失败: userId={}, message={}", user.getUserId(), message);
-            }
-        } catch (Exception e) {
-            log.error("用户同步异常: userId={}, error={}", user.getUserId(), e.getMessage(), e);
-        }
-    }
-
-    /**
-     * 同步用户到 websopy（Map格式，用于MQ消息处理）
-     *
-     * @param data 用户信息Map
-     */
-    public void syncUserToWebsopy(Map<String, Object> data) {
-        String websopyBaseUrl = getWebsopyBaseUrl();
-        if (websopyBaseUrl == null || websopyBaseUrl.isEmpty()) {
-            log.warn("websopyUrl 未配置，跳过用户同步");
-            return;
-        }
-
-        if (data == null || data.isEmpty()) {
-            log.warn("用户数据为空，跳过同步");
-            return;
-        }
-
-        try {
-            Integer userId = getIntValue(data, "userId");
-            if (userId == null) {
-                log.warn("用户数据中缺少userId，跳过同步");
-                return;
-            }
-
-            Integer tenantIdValue = getIntValue(data, "tenantId");
-            if (tenantIdValue == null) {
-                tenantIdValue = 0;
-            }
-
-            // 构建JSON字符串
-            StringBuilder jsonBuilder = new StringBuilder();
-            jsonBuilder.append("{");
-            jsonBuilder.append("\"userId\":").append(userId).append(",");
-            jsonBuilder.append("\"username\":\"").append(escapeJson(getStringValue(data, "username"))).append("\",");
-            jsonBuilder.append("\"nickname\":\"").append(escapeJson(getStringValue(data, "nickname"))).append("\",");
-            jsonBuilder.append("\"avatar\":\"").append(escapeJson(getStringValue(data, "avatar"))).append("\",");
-            jsonBuilder.append("\"phone\":\"").append(escapeJson(getStringValue(data, "phone"))).append("\",");
-            jsonBuilder.append("\"status\":").append(getIntValue(data, "status", 1)).append(",");
-            jsonBuilder.append("\"updateTime\":\"").append(LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"))).append("\",");
-            jsonBuilder.append("\"tenantId\":").append(tenantIdValue).append(",");
-            jsonBuilder.append("\"tenant_id\":").append(tenantIdValue);
-            jsonBuilder.append("}");
-
-            String url = websopyBaseUrl + "/api/app/user-sync/single";
-            String body = jsonBuilder.toString();
-
-            log.info("MQ同步用户到 websopy: userId={}, body={}", userId, body);
-
-            Map<String, String> headers = new HashMap<>();
-            headers.put("Content-Type", "application/json");
-
-            HttpResponse response = HttpUtils.doPost(url, "", "POST", headers, null, body);
-            String responseBody = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
-
-            JSONObject result = JSON.parseObject(responseBody);
-            if (result != null && result.getIntValue("code") == 0) {
-                log.info("MQ用户同步成功: userId={}", userId);
-            } else {
-                String message = result != null ? result.getString("message") : "未知错误";
-                log.error("MQ用户同步失败: userId={}, message={}", userId, message);
-            }
-        } catch (Exception e) {
-            log.error("MQ用户同步异常: error={}", e.getMessage(), e);
-        }
-    }
-
-    /**
-     * 从 websopy 删除用户
-     *
-     * @param data 包含userId的数据
-     */
-    public void deleteUserFromWebsopy(Map<String, Object> data) {
-        String websopyBaseUrl = getWebsopyBaseUrl();
-        if (websopyBaseUrl == null || websopyBaseUrl.isEmpty()) {
-            log.warn("websopyUrl 未配置，跳过删除用户");
-            return;
-        }
-
-        if (data == null) {
-            log.warn("删除用户数据为空");
-            return;
-        }
-
-        try {
-            Integer userId = getIntValue(data, "userId");
-            if (userId == null) {
-                log.warn("删除用户数据中缺少userId");
-                return;
-            }
-
-            String url = websopyBaseUrl + "/api/app/user-sync/delete/" + userId;
-
-            log.info("从 websopy 删除用户: userId={}", userId);
-
-            Map<String, String> headers = new HashMap<>();
-            headers.put("Content-Type", "application/json");
-
-            HttpResponse response = HttpUtils.doPost(url, "", "POST", headers, null, "");
-            String responseBody = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
-
-            JSONObject result = JSON.parseObject(responseBody);
-            if (result != null && result.getIntValue("code") == 0) {
-                log.info("MQ删除用户成功: userId={}", userId);
-            } else {
-                String message = result != null ? result.getString("message") : "未知错误";
-                log.error("MQ删除用户失败: userId={}, message={}", userId, message);
-            }
-        } catch (Exception e) {
-            log.error("MQ删除用户异常: error={}", e.getMessage(), e);
-        }
-    }
-
-    private String getStringValue(Map<String, Object> data, String key) {
-        Object value = data.get(key);
-        return value != null ? String.valueOf(value) : "";
-    }
-
-    private Integer getIntValue(Map<String, Object> data, String key) {
-        return getIntValue(data, key, null);
-    }
-
-    private Integer getIntValue(Map<String, Object> data, String key, Integer defaultValue) {
-        Object value = data.get(key);
-        if (value == null) {
-            return defaultValue;
-        }
-        if (value instanceof Integer) {
-            return (Integer) value;
-        }
-        if (value instanceof Number) {
-            return ((Number) value).intValue();
-        }
-        try {
-            return Integer.parseInt(String.valueOf(value));
-        } catch (NumberFormatException e) {
-            return defaultValue;
-        }
+    @Deprecated
+    public void syncUserToWebsopy(Object user) {
+        log.debug("UserSyncService.syncUserToWebsopy 已废弃，用户同步通过 MQ 自动触发");
     }
 
     /**
      * 刷新 websopy 端的用户缓存
-     * 只传 userId，websopy 端会通过 API 回查获取完整信息
-     *
-     * @param userId 用户ID
+     * 
+     * @deprecated 已废弃
      */
+    @Deprecated
     public void refreshUserCache(Integer userId) {
-        String websopyBaseUrl = getWebsopyBaseUrl();
-        if (websopyBaseUrl == null || websopyBaseUrl.isEmpty()) {
-            log.warn("websopyUrl 未配置，跳过刷新缓存: userId={}", userId);
-            return;
-        }
-
-        if (userId == null) {
-            log.warn("userId 为空，跳过刷新");
-            return;
-        }
-
-        try {
-            String url = websopyBaseUrl + "/api/app/user-sync/refresh/" + userId;
-
-            log.info("刷新用户缓存: userId={}, url={}", userId, url);
-
-            Map<String, String> headers = new HashMap<>();
-            headers.put("Content-Type", "application/json");
-
-            HttpResponse response = HttpUtils.doPost(url, "", "POST", headers, null, "");
-            String responseBody = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8);
-
-            JSONObject result = JSON.parseObject(responseBody);
-            if (result != null && result.getIntValue("code") == 0) {
-                log.info("刷新缓存成功: userId={}", userId);
-            } else {
-                String message = result != null ? result.getString("message") : "未知错误";
-                log.error("刷新缓存失败: userId={}, message={}", userId, message);
-            }
-        } catch (Exception e) {
-            log.error("刷新缓存异常: userId={}, error={}", userId, e.getMessage(), e);
-        }
+        log.debug("UserSyncService.refreshUserCache 已废弃");
     }
 }

src/main/java/com/gxwebsoft/common/system/service/impl/UserServiceImpl.java

@@ -14,7 +14,10 @@ import com.gxwebsoft.common.system.entity.*;
 import com.gxwebsoft.common.system.mapper.UserMapper;
 import com.gxwebsoft.common.system.param.LoginParam;
 import com.gxwebsoft.common.system.param.UserParam;
+import com.gxwebsoft.common.mq.producer.SyncMessageProducer;
 import com.gxwebsoft.common.system.service.*;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.security.core.userdetails.UserDetails;
 import org.springframework.security.core.userdetails.UsernameNotFoundException;
 import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
@@ -34,6 +37,7 @@ import static com.gxwebsoft.common.core.constants.PlatformConstants.WEB;
  * @author WebSoft
  * @since 2018-12-24 16:10:14
  */
+@Slf4j
 @Service
 public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements UserService {
     @Resource
@@ -51,6 +55,9 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
     @Resource
     private UserRefereeService userRefereeService;
 
+    @Autowired(required = false)
+    private SyncMessageProducer syncMessageProducer;
+
     @Override
     public PageResult<User> pageRel(UserParam param) {
         PageParam<User, UserParam> page = new PageParam<>(param);
@@ -130,6 +137,14 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
                 throw new BusinessException("用户角色添加失败");
             }
         }
+        // 用户创建成功后，通过MQ异步同步用户数据到 websopy
+        if (result && syncMessageProducer != null) {
+            User savedUser = getAllByUserId(String.valueOf(user.getUserId()));
+            if (savedUser != null) {
+                syncMessageProducer.sendUserSyncMessage("websopy", "CREATE", savedUser);
+                log.info("用户创建后发送MQ消息同步到websopy: userId={}, phone={}", user.getUserId(), user.getPhone());
+            }
+        }
         return result;
     }
 
@@ -163,6 +178,14 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
                 throw new BusinessException("用户角色添加失败");
             }
         }
+        // 用户更新成功后，通过MQ异步同步用户数据到 websopy
+        if (result && syncMessageProducer != null) {
+            User updatedUser = getAllByUserId(String.valueOf(user.getUserId()));
+            if (updatedUser != null) {
+                syncMessageProducer.sendUserSyncMessage("websopy", "UPDATE", updatedUser);
+                log.info("用户更新后发送MQ消息同步到websopy: userId={}, phone={}", user.getUserId(), user.getPhone());
+            }
+        }
         return result;
     }
 
@@ -323,6 +346,7 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
         // Ensure caller (e.g. register / invite register) gets non-empty roles/authorities in response.
         addUser.setRoles(userRoleService.listByUserId(addUser.getUserId()));
         addUser.setAuthorities(roleMenuService.listMenuByUserId(addUser.getUserId(), null));
+        // addUser内部调用saveUser，saveUser已发送MQ消息，这里不需要重复发送
         return addUser;
     }