第4期：AI 与新模块：snail-ai、MCP、MQTT、ES、Push 如何接入

AI、MCP、MQTT、ES、Push 这些词放在一起，很容易让文章变成概念介绍。但从新老版本对比看，更值得关注的是“它们被放在了哪里、默认是否启用、业务模块通过什么入口接入”。

6.X 没有把所有新能力都强行打开，而是通过独立服务、公共 starter、自动配置和配置开关来接入。比如 snail-ai 是独立服务，MCP 依赖 McpSyncClient Bean，ES 和 MQTT 都是按配置开关启用，Push 则贯穿后端消息体和前端 push.ts。

所以这篇不讨论 AI 能做什么，而是把新增模块当成工程能力来看：哪些是独立服务，哪些是公共模块，哪些依赖外部服务和配置开关。

更新概览

1. snail-ai：新增独立服务

snail-ai 没有被塞进主服务里，而是以独立服务方式出现。它有自己的端口、上下文路径、上传目录和监控认证，是新增模块里边界最清楚的一类。

### 服务配置```yamlserver:  port: 8900  servlet:    context-path: /snail-aispring:  application:    name: ruoyi-snailai-server  web:    resources:      static-locations: classpath:admin/snail-ai:  server:    grpc-port: 18888  skill:    upload-dir: ./upload/skills

启动类

@SpringBootApplicationpublicclassSnailAiServerApplication{publicstaticvoidmain(String[] args){        com.aizuda.snail.ai.starter.SnailAiSpringbootApplication.main(args);    }}

2. snail-ai 依赖接入

<snailai.version>最新版本</snailai.version><!-- Snail AI Agent 启动器 --><dependency><groupId>com.aizuda</groupId><artifactId>snail-ai-agent-starter</artifactId><version>${snailai.version}</version></dependency><!-- Snail AI OpenAPI 启动器 --><dependency><groupId>com.aizuda</groupId><artifactId>snail-ai-openapi-starter</artifactId><version>${snailai.version}</version></dependency>

3. MCP：公共模块自动配置

自动配置

@AutoConfigurationpublicclassMcpAutoConfiguration{@Bean@ConditionalOnBean(McpSyncClient.class)    @ConditionalOnMissingBeanpublicMcpClientTemplatemcpClientTemplate(List<McpSyncClient> mcpSyncClients) {returnnew McpClientTemplate(mcpSyncClients);    }}

查询所有 MCP Server 的工具

public Map<String, List<String>> listTools() {    Map<String, List<String>> result = new LinkedHashMap<>();for (McpSyncClient client : mcpSyncClients) {        List<String> tools = client.listTools().tools().stream()            .map(McpSchema.Tool::name)            .toList();        result.put(getServerName(client), tools);    }return result;}

调用所有 Server 上的同名工具

public Map<String, McpToolCallResult> callTool(String toolName, Map<String, Object> arguments){    Map<String, McpToolCallResult> result = new LinkedHashMap<>();for (McpSyncClient client : mcpSyncClients) {        McpSchema.CallToolResult callResult = client.callTool(new McpSchema.CallToolRequest(toolName, arguments)        );        result.put(getServerName(client), McpToolCallResult.of(getServerName(client), callResult));    }return result;}

调用指定 Server

public Optional<McpToolCallResult> callTool(String serverName,                                           String toolName,                                           Map<String, Object> arguments){return findClient(serverName)        .map(client -> McpToolCallResult.of(            serverName,            client.callTool(new McpSchema.CallToolRequest(toolName, arguments))        ));}

读取资源

public Optional<McpResourceReadResult> readResource(String serverName, String uri){return findClient(serverName)        .map(client -> McpResourceReadResult.of(            serverName,            client.readResource(new McpSchema.ReadResourceRequest(uri))        ));}

4. Elasticsearch：按配置开关启用

自动配置类

@AutoConfiguration@ConditionalOnProperty(value = "easy-es.enable", havingValue = "true")@EsMapperScan("org.dromara.**.esmapper")publicclassEasyEsConfiguration{}

配置示例

# elasticsearch 功能配置# 文档地址: https://www.easy-es.cn/easy-es:# 是否开启EE自动配置enable:false# 兼容模式compatible:true

说明： 默认关闭，打开 easy-es.enable=true 后扫描 org.dromara.**.esmapper。

5. MQTT：默认关闭，开启后自动装配

配置入口

mqtt.client:# 是否开启客户端，默认：trueenabled:false# 连接的服务端 ip ，默认：127.0.0.1ip:127.0.0.1

自动配置

@AutoConfiguration@ConditionalOnProperty(value = "mqtt.client.enabled", havingValue = "true")publicclassMqttAutoConfiguration{@Beanpublic MqttClientConnectListener mqttClientConnectListener(MqttClientCreator mqttClientCreator){returnnew MqttClientConnectListener(mqttClientCreator);    }@Beanpublic MqttClientGlobalMessageListener mqttClientGlobalMessageListener(){returnnew MqttClientGlobalMessageListener();    }}

MQTT 使用虚拟线程

@Beanpublic MqttClientCustomizer mqttClientCustomizer(){return creator -> {int corePoolSize = ThreadUtils.CORE_POOL_SIZE;        ThreadFactory factory = new VirtualThreadTaskExecutor("tio-worker-virtual").getVirtualThreadFactory();        SynThreadPoolExecutor tioExecutor = new SynThreadPoolExecutor(            corePoolSize, corePoolSize, 0L, new LinkedBlockingQueue<>(), factory, new TioCallerRunsPolicy()        );        tioExecutor.prestartCoreThread();        creator.tioExecutor(tioExecutor);    };}

全局消息监听

@Slf4jpublicclassMqttClientGlobalMessageListenerimplementsIMqttClientGlobalMessageListener{@OverridepublicvoidonMessage(ChannelContext context, String topic, MqttPublishMessage message, byte[] payload){        log.info("MqttGlobalMessageEvent => topic: {}, msg: {}",            topic,new String(payload, StandardCharsets.UTF_8));    }}

6. Push：后端结构化消息体

@DatapublicclassPushPayloadDTOimplementsSerializable{private Long messageId;private String type;private String source;private String message;private Object data;private String path;private Long timestamp;publicstatic PushPayloadDTO of(String type, String source, String message, Object data){        PushPayloadDTO payload = new PushPayloadDTO();        payload.setType(StringUtils.defaultIfBlank(type, PushTypeEnum.MESSAGE.getType()));        payload.setSource(StringUtils.defaultIfBlank(source, PushSourceEnum.BACKEND.getSource()));        payload.setMessage(message);        payload.setData(data);        payload.setTimestamp(System.currentTimeMillis());return payload;    }}

带跳转路径

publicstatic PushPayloadDTO of(PushTypeEnum type,                                PushSourceEnum source,                                String message,                                Object data,                                String path){    PushPayloadDTO payload = of(type, source, message, data);    payload.setPath(path);return payload;}

登录后推送示例

messageService.publishMessage(    List.of(userId),    PushPayloadDTO.of(        PushTypeEnum.MESSAGE,        PushSourceEnum.BACKEND,        DateUtils.getTodayHour(new Date()) + "好，欢迎登录 RuoYi-Vue-Plus 后台管理系统",null    ));

7. 前端 Push 对接

exportconst initPush = () => {  closePush();if (import.meta.env.VITE_APP_MESSAGE_ENABLED === 'false') {return;  }const path = import.meta.env.VITE_APP_MESSAGE_PATH || '/resource/message';const transport = import.meta.env.VITE_APP_MESSAGE_TRANSPORT || 'sse';if (transport.toLowerCase() === 'websocket') {    initWsPush(buildWsUrl(path));return;  }  initSsePush(buildSseUrl(path));};

统一清理连接

exportconst closePush = () => {  closePushConnection?.();  closePushConnection = undefined;  stopPushWatchers.forEach(stop => stop());  stopPushWatchers = [];};

写在最后

这几类新增能力有一个共同特点：入口都很清楚，同时也和主链路保持了边界。snail-ai 涉及独立服务和端口，ES 依赖集群配置，MQTT 依赖 Broker，MCP 依赖实际的 Server 和 Client Bean。

从模块设计看，新版更倾向于“能力可插拔”：独立服务单独运行，公共模块通过自动配置接入，ES、MQTT 这类外部依赖默认由配置开关控制。这样既能把新能力放进项目体系，又不会让主链路被所有扩展能力强绑定。