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

Stream Key: user:actions:stream
    |
    ├── 消息ID (格式: timestamp-sequence)
    |   ├── field1: value1
    |   ├── field2: value2
    |   └── field3: value3
    |
    ├── 消息ID
    |   ├── field1: value1
    |   └── ...
    |
    └── 消费者组 (consumer groups)
        ├── group1 (analyze-group)
        |   ├── consumer1 (analyze-consumer-xxx)
        |   └── consumer2 (analyze-consumer-yyy)
        |
        └── group2 (vectorize-group)
            ├── consumer1 (vectorize-consumer-xxx)
            └── ...

1
2
3
4
5

<dependency>
    <groupId>org.redisson</groupId>
    <artifactId>redisson-spring-boot-starter</artifactId>
    <version>4.0.0</version>
</dependency>

 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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

@Service
@Slf4j
@RequiredArgsConstructor
public class RedisService {

    private final RedissonClient redissonClient;

    /**
     * 发送消息到 Stream
     */
    public String streamAdd(String streamKey, Map<String, String> message, int maxLen) {
        RStream<String, String> stream = redissonClient.getStream(streamKey, StringCodec.INSTANCE);
        StreamAddArgs<String, String> addArgs = StreamAddArgs.entries(message);

        if (maxLen > 0) {
            // 非严格裁剪，超过最大长度时裁剪旧消息
            addArgs.trimNonStrict().maxLen(maxLen);
        }
        StreamMessageId streamMessageId = stream.add(addArgs);
        return streamMessageId.toString();
    }

    /**
     * 创建消费者组
     */
    public void createConsumer(String streamKey, String groupName) {
        RStream<Object, Object> stream = redissonClient.getStream(streamKey, StringCodec.INSTANCE);
        try {
            stream.createGroup(StreamCreateGroupArgs.name(groupName).makeStream());
        } catch (Exception e) {
            // 忽略组已存在的异常
            if (!e.getMessage().contains("BUSYGROUP")) {
                log.warn("创建消费者组失败: {}", e.getMessage());
            }
        }
    }

    /**
     * 消费消息（阻塞模式）
     */
    public boolean streamConsumeMessages(String streamKey, String groupName,
            String consumerName, int count, long blockTimeoutMs,
            StreamMessageProcessor processor) {
        RStream<String, String> stream = redissonClient.getStream(streamKey, StringCodec.INSTANCE);

        try {
            Map<StreamMessageId, Map<String, String>> messages = stream.readGroup(
                groupName, consumerName,
                StreamReadGroupArgs.neverDelivered().count(count)
                    .timeout(Duration.ofMillis(blockTimeoutMs)));

            if (messages == null || messages.isEmpty()) {
                return false;
            }

            for (StreamMessageId streamMessageId : messages.keySet()) {
                processor.process(streamMessageId, messages.get(streamMessageId));
            }
            return true;
        } catch (Exception e) {
            // Redisson bug: 无消息时返回 EmptyList
            return false;
        }
    }

    /**
     * 确认消息
     */
    public void streamAck(String streamKey, String groupName, StreamMessageId... ids) {
        RStream<Object, Object> stream = redissonClient.getStream(streamKey, StringCodec.INSTANCE);
        stream.ack(groupName, ids);
    }

    @FunctionalInterface
    public interface StreamMessageProcessor {
        void process(StreamMessageId messageId, Map<String, String> data);
    }
}

 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
29
30
31
32

@Slf4j
public abstract class AbstractStreamProducer<T> {

    private final RedisService redisService;

    protected AbstractStreamProducer(RedisService redisService) {
        this.redisService = redisService;
    }

    protected String sendTask(T data) {
        try {
            String streamMessageId = redisService.streamAdd(
                streamKey(),
                buildMessage(data),
                AsyncTaskStreamConstants.BATCH_SIZE
            );
            log.info("{}任务已发送到Stream: {}, messageId={}",
                taskDisplayName(), payloadIdentifier(data), streamMessageId);
            return streamMessageId;
        } catch (Exception e) {
            log.error("发送{}任务失败: {}", taskDisplayName(), e.getMessage(), e);
            onSendFailed(data, "任务入队失败: " + e.getMessage());
        }
        return "";
    }

    protected abstract String taskDisplayName();
    protected abstract String streamKey();
    protected abstract Map<String, String> buildMessage(T payload);
    protected abstract String payloadIdentifier(T payload);
    protected abstract void onSendFailed(T payload, String error);
}

 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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

@Service
@Slf4j
public class AnalyzeStreamProducer
        extends AbstractStreamProducer<AnalyzeStreamProducer.AnalyzeTaskPayload> {

    private final ResumeRepository resumeRepository;
    private final ObjectMapper objectMapper;

    protected AnalyzeStreamProducer(RedisService redisService,
            ResumeRepository resumeRepository, ObjectMapper objectMapper) {
        super(redisService);
        this.resumeRepository = resumeRepository;
        this.objectMapper = objectMapper;
    }

    public record AnalyzeTaskPayload(Long resumeId, String content) {}

    public String analyzeStreamSend(Long resumeId, String content) {
        return sendTask(new AnalyzeTaskPayload(resumeId, content));
    }

    @Override
    protected String taskDisplayName() {
        return "简历分析";
    }

    @Override
    protected String streamKey() {
        return AsyncTaskStreamConstants.RESUME_ANALYZE_STREAM_KEY;
    }

    @Override
    protected Map<String, String> buildMessage(AnalyzeTaskPayload payload) {
        return Map.of(
            AsyncTaskStreamConstants.FIELD_RESUME_ID, payload.resumeId().toString(),
            AsyncTaskStreamConstants.FIELD_CONTENT, payload.content(),
            AsyncTaskStreamConstants.FIELD_RETRY_COUNT, "0"
        );
    }

    @Override
    protected String payloadIdentifier(AnalyzeTaskPayload payload) {
        return objectMapper.writeValueAsString(payload);
    }

    @Override
    protected void onSendFailed(AnalyzeTaskPayload payload, String error) {
        updateAnalyzeStatus(payload.resumeId, AsyncTaskStatus.FAILED,
            truncateError(error));
    }
}

 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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98

@Slf4j
public abstract class AbstractStreamConsumer<T> {

    private final RedisService redisService;
    private ExecutorService executorService;
    private final AtomicBoolean running = new AtomicBoolean(false);
    private String consumerName;

    protected AbstractStreamConsumer(RedisService redisService) {
        this.redisService = redisService;
    }

    @PostConstruct
    public void init() {
        // 创建消费者组
        this.consumerName = consumerPrefix() + UUID.randomUUID().toString().substring(0, 8);
        try {
            redisService.createConsumer(streamKey(), groupName());
        } catch (Exception e) {
            log.error("创建消费者组发生异常（可能已存在）: {}", e.getMessage());
        }

        // 创建单线程执行器
        this.executorService = new ThreadPoolExecutor(1, 1, 0L,
            TimeUnit.MILLISECONDS, new LinkedBlockingDeque<>(),
            r -> {
                Thread t = new Thread(r, threadName());
                t.setDaemon(true);
                return t;
            });

        running.set(true);
        executorService.submit(this::consumeLoop);
    }

    @PreDestroy
    public void destroy() {
        running.set(false);
        if (executorService != null) {
            executorService.shutdown();
        }
    }

    private void consumeLoop() {
        while (running.get()) {
            try {
                redisService.streamConsumeMessages(
                    streamKey(), groupName(), consumerName,
                    AsyncTaskStreamConstants.BATCH_SIZE,
                    AsyncTaskStreamConstants.POLL_INTERVAL_MS,
                    this::processMessage
                );
            } catch (Exception e) {
                log.error("消费消息时发生错误: {}", e.getMessage(), e);
            }
        }
    }

    private void processMessage(StreamMessageId streamMessageId, Map<String, String> data) {
        T message = parsePayload(streamMessageId, data);
        if (isAck(message)) {
            ackMessage(streamMessageId);
        }

        int retryCount = getRetryCount(data);

        try {
            markProcessing(message);
            processBusiness(message);
            markCompleted(message);
            ackMessage(streamMessageId);
        } catch (Exception e) {
            if (retryCount < MAX_RETRY_COUNT) {
                retryMessage(message, retryCount);
            } else {
                markFailed(message, truncateError(
                    taskDisplayName() + "失败(已重试" + retryCount + "次): " + e.getMessage()
                ));
            }
            ackMessage(streamMessageId);
        }
    }

    protected abstract void retryMessage(T payload, int retryCount);
    protected abstract String taskDisplayName();
    protected abstract void markFailed(T message, String errorMsg);
    protected abstract void markCompleted(T message);
    protected abstract void processBusiness(T message);
    protected abstract void markProcessing(T message);
    protected abstract String payloadIdentifier(T message);
    protected abstract int getRetryCount(Map<String, String> data);
    protected abstract T parsePayload(StreamMessageId streamMessageId,
                                       Map<String, String> data);
    protected abstract String groupName();
    protected abstract String consumerPrefix();
    protected abstract String threadName();
    protected abstract String streamKey();
}

 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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

@Service
@Slf4j
public class AnalyzeStreamConsumer
        extends AbstractStreamConsumer<AnalyzeStreamConsumer.AnalyzePayload> {

    private final ResumeRepository resumeRepository;
    private final ResumePersistenceService persistenceService;
    private final ResumeGradingService gradingService;

    record AnalyzePayload(Long resumeId, String content) {}

    protected AnalyzeStreamConsumer(RedisService redisService,
            ResumeRepository resumeRepository,
            ResumePersistenceService persistenceService,
            ResumeGradingService gradingService) {
        super(redisService);
        this.resumeRepository = resumeRepository;
        this.persistenceService = persistenceService;
        this.gradingService = gradingService;
    }

    @Override
    protected void retryMessage(AnalyzePayload payload, int retryCount) {
        redisService.streamAdd(streamKey(),
            Map.of(
                FIELD_RESUME_ID, payload.resumeId().toString(),
                FIELD_CONTENT, payload.content(),
                FIELD_RETRY_COUNT, String.valueOf(retryCount)
            ),
            STREAM_MAX_LEN
        );
    }

    @Override
    protected void processBusiness(AnalyzePayload payload) {
        Long resumeId = payload.resumeId();
        if (!resumeRepository.existsById(resumeId)) {
            log.warn("简历已被删除，跳过分析任务: resumeId={}", resumeId);
            return;
        }

        ResumeAnalysisResponse analysis = gradingService.analyzeResume(payload.content());
        ResumeEntity resume = resumeRepository.findById(resumeId).orElse(null);
        if (resume == null) {
            log.warn("简历在分析期间被删除，跳过保存结果: resumeId={}", resumeId);
            return;
        }
        persistenceService.saveAnalysis(resume, analysis);
    }

    @Override
    protected int getRetryCount(Map<String, String> data) {
        return Integer.parseInt(
            data.getOrDefault(FIELD_RETRY_COUNT, "0").toString()
        );
    }

    @Override
    protected AnalyzePayload parsePayload(StreamMessageId streamMessageId,
                                          Map<String, String> data) {
        String resumeIdStr = data.get(FIELD_RESUME_ID);
        String content = data.get(FIELD_CONTENT);
        if (resumeIdStr == null || content == null) {
            log.warn("消息格式错误，跳过: messageId={}", streamMessageId);
            return null;
        }
        return new AnalyzePayload(Long.parseLong(resumeIdStr), content);
    }

    // ... 其他抽象方法实现
}

 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
29
30

public final class AsyncTaskStreamConstants {

    // 通用配置
    public static final int MAX_RETRY_COUNT = 3;
    public static final int BATCH_SIZE = 10;
    public static final long POLL_INTERVAL_MS = 1000;
    public static final int STREAM_MAX_LEN = 1000;

    // 字段名
    public static final String FIELD_RETRY_COUNT = "retryCount";
    public static final String FIELD_CONTENT = "content";
    public static final String FIELD_RESUME_ID = "resumeId";

    // Stream 配置
    public static final String RESUME_ANALYZE_STREAM_KEY = "resume:analyze:stream";
    public static final String RESUME_ANALYZE_GROUP_NAME = "analyze-group";
    public static final String RESUME_ANALYZE_CONSUMER_PREFIX = "analyze-consumer-";

    // 知识库 Stream 配置
    public static final String KB_VECTORIZE_STREAM_KEY = "knowledgebase:vectorize:stream";
    public static final String KB_VECTORIZE_GROUP_NAME = "vectorize-group";

    // 面试评估 Stream 配置
    public static final String INTERVIEW_EVALUATE_STREAM_KEY = "interview:evaluate:stream";
    public static final String INTERVIEW_EVALUATE_GROUP_NAME = "evaluate-group";

    // 语音面试 Stream 配置
    public static final String VOICE_EVALUATE_STREAM_KEY = "voice:evaluate:stream";
    public static final String VOICE_EVALUATE_GROUP_NAME = "voice-evaluate-group";
}

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

@Service
@RequiredArgsConstructor
public class ResumerUploadService {

    private final AnalyzeStreamProducer analyzeStreamProducer;

    public Map<String, Object> uploadAndAnalyze(MultipartFile file) {
        // ... 文件验证、解析、存储逻辑

        // 保存简历到数据库（状态为 PENDING）
        ResumeEntity resume = persistenceService.saveResume(file, resumeText, fileKey, fileUrl);

        // 发送分析任务到 Redis Stream
        analyzeStreamProducer.analyzeStreamSend(resume.getId(), resumeText);

        return Map.of(
            "resume", Map.of(
                "id", resume.getId(),
                "analyzeStatus", AsyncTaskStatus.PENDING.name()
            )
        );
    }
}

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

# 查看 Stream 信息
XINFO STREAM resume:analyze:stream

# 查看消费者组
XINFO GROUPS resume:analyze:stream

# 查看待确认消息 (PEL)
XINFO CONSUMERS resume:analyze:stream analyze-group

# 读取所有消息（不消费）
XRANGE resume:analyze:stream - +

# 手动确认消息
XACK resume:analyze:stream analyze-group 1689123456788-0

# 删除消息
XDEL resume:analyze:stream 1689123456788-0

# 裁剪 Stream 长度
XTRIM resume:analyze:stream MAXLEN 1000

1
2
3
4

# 启动多个消费者进程
java -jar app.jar --consumer-id=1 &
java -jar app.jar --consumer-id=2 &
java -jar app.jar --consumer-id=3 &

1
2
3
4

public long getPendingCount(String streamKey, String groupName) {
    RStream<Object, Object> stream = redissonClient.getStream(streamKey);
    return stream.getGroupInfo(groupName).getPending();
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

protected void markFailed(T message, String errorMsg) {
    // 写入死信队列
    redisService.streamAdd("dlq:" + streamKey(),
        Map.of(
            "originalKey", streamKey(),
            "error", errorMsg,
            "payload", JSON.toJSONString(message),
            "failedAt", LocalDateTime.now().toString()
        )
    );
}

1
2
3

// 平衡延迟和性能
private static final int BATCH_SIZE = 10;
private static final long POLL_INTERVAL_MS = 1000;

1

streamAdd(streamKey, message, STREAM_MAX_LEN); // 最大 1000 条

1
2
3
4
5

@PreDestroy
public void destroy() {
    running.set(false);
    executorService.shutdown();
}

1
2
3
4
5
6
7
8

protected void processBusiness(AnalyzePayload payload) {
    // 检查是否已处理（幂等）
    if (resume.getAnalyzeStatus() == AsyncTaskStatus.COMPLETED) {
        log.info("简历已分析过，跳过: resumeId={}", payload.resumeId());
        return;
    }
    // ... 正常处理逻辑
}