Java应用间Webhook通信的可靠性保障：无需新增基础设施的重试方案

CREATE TABLE outbound_tasks (
    task_id VARCHAR(255) PRIMARY KEY,
    payload TEXT NOT NULL,
    call_status VARCHAR(50) NOT NULL,
    last_retry_ts TIMESTAMP,
    retry_count INT DEFAULT 0,
    created_ts TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.UUID;

@Service
public class OutboundTaskService {

    private final OutboundTaskRepository repository; // 假设有一个JPA Repository

    public OutboundTaskService(OutboundTaskRepository repository) {
        this.repository = repository;
    }

    @Transactional
    public void recordAndAttemptSend(String payload) {
        OutboundTask task = new OutboundTask();
        task.setTaskId(UUID.randomUUID().toString());
        task.setPayload(payload);
        task.setCallStatus("NOT_CALLED");
        task.setCreatedTs(LocalDateTime.now());
        task.setRetryCount(0);
        // last_retry_ts 可以在这里设置为null或当前时间，取决于重试器逻辑

        repository.save(task); // 保存到数据库

        // 首次尝试发送可以立即进行，也可以完全由重试器处理
        // 这里为了简化，假设首次发送也由重试器处理，或者直接在这里尝试发送，失败则更新状态
        // 例如：
        // try {
        //     appAClient.send(payload);
        //     task.setCallStatus("COMPLETE");
        // } catch (Exception e) {
        //     task.setCallStatus("FAILED");
        //     task.setLastRetryTs(LocalDateTime.now());
        // }
        // repository.save(task);
    }
}

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.util.List;

@Component
public class OutboundTaskRetryScheduler {

    private final OutboundTaskRepository repository;
    private final AppAClient appAClient; // 假设是调用App A的REST客户端

    // 配置参数
    private static final int MAX_RETRIES = 5;
    private static final long RETRY_INTERVAL_SECONDS = 30; // 初始重试间隔

    public OutboundTaskRetryScheduler(OutboundTaskRepository repository, AppAClient appAClient) {
        this.repository = repository;
        this.appAClient = appAClient;
    }

    @Scheduled(fixedDelay = 10000) // 每10秒执行一次
    @Transactional
    public void retryFailedTasks() {
        // 查询所有未完成且需要重试的任务
        List<OutboundTask> tasksToRetry = repository.findByCallStatusInAndLastRetryTsBeforeOrLastRetryTsIsNull(
                List.of("NOT_CALLED", "FAILED"), LocalDateTime.now().minusSeconds(RETRY_INTERVAL_SECONDS));

        for (OutboundTask task : tasksToRetry) {
            if (task.getRetryCount() >= MAX_RETRIES) {
                // 达到最大重试次数，标记为最终失败，可能需要人工介入
                task.setCallStatus("FAILED");
                repository.save(task);
                continue;
            }

            task.setCallStatus("WIP"); // 标记为处理中
            task.setLastRetryTs(LocalDateTime.now());
            task.setRetryCount(task.getRetryCount() + 1);
            repository.save(task); // 更新状态到数据库

            try {
                appAClient.send(task.getPayload()); // 尝试发送
                task.setCallStatus("COMPLETE"); // 发送成功
            } catch (Exception e) {
                // 发送失败，保持FAILED状态，等待下次重试
                task.setCallStatus("FAILED");
                // 可以在这里记录具体的错误信息
                System.err.println("Failed to send task " + task.getTaskId() + ": " + e.getMessage());
            } finally {
                repository.save(task); // 最终更新任务状态
            }
        }
    }
}