OCR 异步化 + WebSocket 推送：订单识别的完整链路

拾物清单有个核心功能：新增物品时上传一张电商订单截图，AI 自动识别商品名称、实付金额、购买日期、平台、分类等字段回填表单。第一版实现很直接——Controller 里同步调 DashScope 多模态接口，等 AI 返回 JSON 再响应。结果就是用户全屏遮罩干等 5~20 秒，AI 一抖动整个页面就卡死。

这次改造把链路升级为任务化异步处理 + WebSocket 实时推送，本文按”为什么改 → 怎么设计 → 怎么实现 → 踩了什么坑”完整记录一遍。

一、原实现的痛点

原链路就是最朴素的同步调用：上传图片 → 调 DashScope → 等 5~20 秒 → 返回结果。用起来暴露了一串问题：

二、方案选型：为什么不上消息队列

标准答案肯定是 MQ：任务投递到队列，消费者异步处理，结果事件广播。但服务器就 1.8GB 内存，部署环境不允许再加 RocketMQ/RabbitMQ 这种中间件，运维成本也压不住。

最终选择的组合是：MySQL 任务表 + 线程池 + WebSocket + 轮询兜底。

• 任务表：承担消息队列的持久化语义，天然支持重启恢复、失败重试、轮询查询；
• 线程池：承担异步执行，零额外依赖，单机单实例足够；
• CAS 更新：保证同一任务只被一个 Worker 执行，未来多实例也成立；
• WebSocket：实时推送，体验好；
• 轮询：兜底通道，弱网/代理不支持 WS 时结果仍可达。

代码按”提交/消费”边界组织，流量真上来了把提交层换成 MQ 就行，任务表可以退化为消费记录。

三、数据模型：一张任务表撑起整个状态机

新增 ocr_task 表，字段不多：

1
2
3
4
5
6
7
8
9
10
11
12
13

CREATE TABLE `ocr_task` (
  `task_id`       BIGINT PRIMARY KEY COMMENT '雪花ID，序列化为String防前端精度丢失',
  `user_id`       BIGINT NOT NULL COMMENT '归属用户，业务隔离与WS推送路由依据',
  `status`        VARCHAR(16) NOT NULL COMMENT 'PENDING/PROCESSING/SUCCESS/FAILED',
  `screenshot_url` VARCHAR(512) COMMENT '订单截图OSS地址',
  `result_json`   TEXT COMMENT '识别结果JSON',
  `error_message` VARCHAR(500) COMMENT '最终失败原因',
  `retry_count`   INT DEFAULT 0,
  `create_time`   DATETIME,
  `update_time`   DATETIME COMMENT '兼作僵死任务判定',
  `deleted`       TINYINT DEFAULT 0,
  KEY `idx_user_status` (`user_id`, `status`)
);

状态机就四个状态：

1
2

PENDING → PROCESSING → SUCCESS
                   ↘ FAILED →（手动重试）→ PENDING

任务表在这里承担了五个角色：异步解耦的持久化载体、断点恢复的数据基础、重试的依据、轮询查询的数据源、CAS 防重的条件列。一张表干了 MQ + 延迟队列 + 死信队列的活，单机场景下够用了。

四、后端实现

4.1 上传即返回

1
2
3
4
5

@PostMapping("/recognize")
@RateLimit(capacity = 5, refillPerSecond = 0.2, message = "识别请求过于频繁，请稍后再试")
public Result<OcrTaskVO> recognize(@RequestParam("file") MultipartFile file) {
    return Result.success(ocrService.createTask(file));
}

Service 层三步走：传 OSS → 插任务行（PENDING）→ 提交异步，立即返回 taskId：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

String ossUrl = ossUtil.upload(file, "ocr");

OcrTask task = new OcrTask();
task.setUserId(UserContext.getUserId());
task.setStatus(OcrTask.STATUS_PENDING);
task.setScreenshotUrl(ossUrl);
taskMapper.insert(task);

try {
    ocrTaskProcessor.submit(task.getTaskId());   // @Async，跨 Bean 调用代理才生效
} catch (Exception e) {
    // 线程池队列满（TaskRejectedException）→ 标记 FAILED，前端可手动重试
    log.warn("OCR 任务提交失败 taskId={}: {}", task.getTaskId(), e.getMessage());
    // ...置 FAILED("系统繁忙，请稍后重试")
}

线程池配置刻意用了 AbortPolicy 而不是 CallerRunsPolicy——队列满时回退到调用线程执行，等于把压力反压回 HTTP 线程，异步就白做了。拒绝时直接标失败，让用户稍后手动重试，链路更可控。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

@Configuration
@EnableAsync
public class AsyncConfig {
    @Bean("ocrExecutor")
    public ThreadPoolTaskExecutor ocrExecutor(OcrProperties properties) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(properties.getExecutor().getCorePoolSize());    // 2
        executor.setMaxPoolSize(properties.getExecutor().getMaxPoolSize());      // 4
        executor.setQueueCapacity(properties.getExecutor().getQueueCapacity());  // 50
        executor.setThreadNamePrefix("ocr-worker-");
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());
        executor.initialize();
        return executor;
    }
}

参数 core=2 / max=4 / queue=50 是照着 1.8GB 内存的服务器定的——AI 调用是 IO 密集型，线程多了也只是一起挂着等网络。

4.2 CAS 抢防重复识别

Worker 拿到任务后，第一件事是用数据库 CAS 抢占：

1
2

UPDATE ocr_task SET status = 'PROCESSING', update_time = NOW()
WHERE task_id = #{taskId} AND status = 'PENDING'

影响行数 = 1 抢占成功继续干活；= 0 说明已被别人接管直接返回。比”先查状态再更新”安全得多，而且天然支持多实例——将来水平扩展也不用引分布式锁。

4.3 自动重试 + 指数退避

1
2
3
4
5
6
7
8
9
10
11
12

for (int attempt = 1; attempt <= maxAttempts; attempt++) {
    try {
        result = recognize(task.getScreenshotUrl());
        break;
    } catch (Exception e) {
        errorMessage = e.getMessage();
        taskMapper.update(...retryCount = attempt...);   // 记录重试痕迹
        if (attempt < maxAttempts) {
            Thread.sleep(attempt * 1000L);               // 1s、2s 线性退避
        }
    }
}

默认最多 3 次。注意异步线程里没有请求上下文，UserContext（ThreadLocal）要手动注入、finally 里清掉，否则线程复用会串用户：

1
2
3
4
5
6

UserContext.setUserId(task.getUserId());
try {
    doProcess(task);
} finally {
    UserContext.remove();
}

4.4 重启恢复

OcrTaskRecoveryRunner 实现 ApplicationRunner，启动时扫 PENDING 或超时 5 分钟的 PROCESSING 任务（LIMIT 20，防止启动瞬间打爆 AI 接口），逐个重新入队。配合 CAS，”假死的慢任务”和”恢复的任务”不会重复识别。

五、WebSocket 实时推送

5.1 握手鉴权：token 只能放 query

H5 的浏览器 WebSocket API 不能自定义 Header，所以 token 走 query 参数，由握手拦截器校验：

1
2
3
4
5
6
7
8
9

public boolean beforeHandshake(ServerHttpRequest request, ...,
                               Map<String, Object> attributes) {
    String token = servletRequest.getServletRequest().getParameter("token");
    if (!StringUtils.hasText(token) || !jwtUtil.validateToken(token)) {
        return false;   // 拒绝握手
    }
    attributes.put("userId", jwtUtil.getUserIdFromToken(token));
    return true;
}

安全上的已知妥协：token 会出现在访问日志里，生产要配 HTTPS，更严谨的做法是握手后换短期一次性票据。另外 JWT 过期后已建立的连接仍然存活，后续可以做心跳时复验。

同时记得把 /api/ws 从 JWT 拦截器的放行清单里处理好——握手拦截器和 MVC 拦截器是两套，别重复校验也别双双漏掉。

5.2 会话管理与推送

会话容器按 userId 维度管理，支持同一用户多设备同时在线：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

private static final ConcurrentMap<Long, CopyOnWriteArraySet<WebSocketSession>> SESSIONS =
        new ConcurrentHashMap<>();

public void pushTaskUpdate(Long userId, OcrTaskVO vo) {
    CopyOnWriteArraySet<WebSocketSession> set = SESSIONS.get(userId);
    if (set == null || set.isEmpty()) return;
    String json = objectMapper.writeValueAsString(vo);
    for (WebSocketSession session : set) {
        if (!session.isOpen()) continue;
        try {
            session.sendMessage(new TextMessage(json));
        } catch (IOException e) {
            log.warn("推送消息失败 session={}: {}", session.getId(), e.getMessage());
        }
    }
}

选型说明：ConcurrentHashMap 保证并发安全；CopyOnWriteArraySet 适合”读远多于写”的场景——推送时遍历是读，连接开关是写，一个用户也就一两个连接，写时复制开销可忽略。连接关闭时在 afterConnectionClosed 里移除会话，空集合也删掉，不会泄漏。

心跳很简单：客户端每 30 秒发 ping，服务端回 pong，防止中间设备掐掉空闲连接。

六、前端：WS + 轮询双通道

uni-app 用 uni.connectSocket 拿到 socketTask，封装成单例模块 utils/websocket.js。核心是这个 waitOcrTask：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

export function waitOcrTask(taskId, { timeout = 60000 } = {}) {
    return new Promise((resolve, reject) => {
        let settled = false
        const succeed = (result) => {
            if (settled) return
            settled = true; cleanup(); resolve(result)
        }
        const fail = (message) => {
            if (settled) return
            settled = true; cleanup(); reject(new Error(message || '识别失败'))
        }
        const handleUpdate = (payload) => {
            if (payload.status === 'SUCCESS') succeed(payload.result)
            else if (payload.status === 'FAILED') fail(payload.errorMessage)
        }

        unsubscribe = subscribeTask(taskId, handleUpdate)   // 通道一：WS 实时推送

        pollTimer = setInterval(async () => {               // 通道二：3 秒轮询兜底
            try {
                const data = await get(`/ocr/tasks/${taskId}`)
                handleUpdate(data)
            } catch (e) { /* 网络抖动忽略，继续轮询 */ }
        }, 3000)

        timeoutTimer = setTimeout(() => fail('识别超时，请重试'), timeout)
    })
}

两个通道谁先到谁生效，settled 标志保证只处理一次，重复消息无害。WS 挂了、代理不支持升级、弱网断连——轮询都能把结果捞回来。

断线重连用指数退避，1s → 2s → 4s… 上限 10s，且只在有活跃订阅时才重连，避免页面闲置后还在后台无限重试：

1
2
3
4
5
6

function scheduleReconnect() {
    if (reconnectTimer || manualClosed) return
    reconnectAttempts += 1
    const delay = Math.min(1000 * Math.pow(2, reconnectAttempts), 10000)
    reconnectTimer = setTimeout(() => { ... connect() }, delay)
}

交互层面，上传后拿到 taskId 就把全屏遮罩换成表单内的进度提示，识别期间用户可以继续填别的字段；完成时 applyOcrResult(result) 自动回填，失败弹窗给”重试”按钮直接走 POST /api/ocr/tasks/{taskId}/retry。

七、闭环：识别记录页

做了异步就自然冒出一个新问题：用户上传后直接退出编辑页，任务在后台完成了，结果去哪看？

补了两个接口（GET /api/ocr/tasks 分页列表 + DELETE /api/ocr/tasks/{taskId} 放弃识别），首页展示”待处理识别”横幅，点击进识别记录页，点任意一条复用 edit-item.vue 的 review 模式核对。保存时 GoodsSaveDTO.ocrTaskId 随物品一起提交，同一个事务里插物品 + 逻辑删除任务，杜绝重复保存。识别 → 核对 → 保存，业务上 AI 只辅助填写，不直接生成物品。

八、踩坑记录

1、@Async 自调用不生效

@Async 基于 Spring AOP 代理，同类内 this.submit() 不走代理，会退化成同步执行。必须跨 Bean 调用（OcrServiceImpl 调 OcrTaskProcessor.submit()）。一开始写在同一个类里，异步压根没触发，线程名还是 http-nio-8080-exec-*，查了半晌。

2、H5 联调 WS 不通

HBuilderX 的 devServer 代理默认不开 WebSocket 升级，manifest.json 里 /api 代理必须加 ws: true，否则 H5 环境永远连不上，APP 真机却一切正常——平台差异型坑。

3、测试上下文撞车

集成测试里 @MockBean 会创建第二个 Spring 上下文，和主上下文共享同一个 H2 内存库（DB_CLOSE_DELAY=-1），表被重复初始化直接主键冲突。解决：给测试类配独立库名 jdbc:h2:mem:ocrtask。另外 Mockito 里 thenThrow 之后再 when() 重打桩会先触发旧桩，改用 doReturn/doThrow 风格更稳。

4、BigDecimal 精度丢失

result_json 用 Hutool 序列化，BigDecimal("6999.00") 往返后变 6999，scale 丢了。表单回填无感知，但哪天要做精确展示就会踩上。根因是 DTO 层用了字符串透传、VO 层才转 BigDecimal，中间那一次 JSON 往返丢了标度。

5、雪花 ID 精度

项目全局把 Long 序列化成 String，任务 taskId 也是。前端如果按 Number 处理 taskId，比较和作为对象 key 都会出问题，订阅时统一 String(taskId) 归一化。

九、测试与遗留

4 个集成用例覆盖主链路：创建后自动识别成功、连续失败 3 次再重试成功、他人任务不可查询、非 FAILED 不可重试。全量 mvn test 通过。

明确知道但还没做的：

1. 孤儿 OSS 清理：失败/放弃任务的截图没删（识别失败时还留着给重试用），需要定时任务兜底；
2. 单任务总超时：慢调用会长占线程，应加 60s 级别的硬超时；
3. 多实例广播：WS 会话是内存态，用户连实例 A、任务在实例 B 完成时 A 收不到，要上 Redis Pub/Sub 或 MQ 广播——这是当前单实例部署的已知演进边界；
4. 消息队列演进：流量增大后把提交层换成 RocketMQ，任务表转为消费记录。

写在最后

这套”任务表 + 线程池 + WS + 轮询”的组合算不上什么高大上的架构，但它在零中间件、小内存的约束下把不阻塞、不丢失、可重试、能感知四个目标都兜住了。工程上大部分问题都不是”选哪个最先进的技术”，而是”在约束条件下把每条边界都封上”。MQ、租约、分布式广播都在演进清单里，等真有那个量级再上不迟。