使用Java Gmail API批量请求遇429错误，如何实现重试？

官方明确说明，批量请求中使用HttpBackOffUnsuccessfulResponseHandler会导致每个失败请求单独触发退避策略，反而可能加剧限流问题。针对你的场景，推荐以下几种处理方式：

1. 收集失败请求，单独发起带退避的重试 #

在onFailure回调里记录触发429限流的消息ID，等批量请求执行完毕后，对这些失败ID单独发起请求，并给单个请求配置退避策略。

修改后的示例代码：

List<MessageDto> messages = new ArrayList<>();
String nextPageToken = null;
// 收集触发限流的消息ID
List<String> failedMessageIds = new ArrayList<>();

List<Message> batchMessages = new ArrayList<>();
ListMessagesResponse response =
        gmailClient(googleId)
                .users()
                .messages()
                .list(googleId)
                .setQ(query)
                .setPageToken(pageToken)
                .setMaxResults(maxResults == null ? 50 : maxResults)
                .execute();

if (response.getMessages() != null) {
    batchMessages.addAll(response.getMessages());
    nextPageToken = response.getNextPageToken();

    BatchRequest batch = gmailClient(googleId).batch();

    // 遍历消息时绑定ID到回调，方便失败时记录
    for (Message message : batchMessages) {
        final String msgId = message.getId();
        gmailClient(googleId)
                .users()
                .messages()
                .get(googleId, msgId)
                .setFormat("raw")
                .queue(batch, new JsonBatchCallback<Message>() {
                    @Override
                    public void onSuccess(Message message, HttpHeaders responseHeaders) {
                        messages.add(MessageMapper.getMessage(googleId, message, false));
                    }

                    @Override
                    public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
                        log.error("处理消息 {} 失败: {}", msgId, e.getMessage());
                        // 仅收集429限流错误的ID
                        if (e.getCode() == 429) {
                            failedMessageIds.add(msgId);
                        }
                    }
                });
    }
    batch.execute();

    // 对失败的消息单独发起带退避的重试
    if (!failedMessageIds.isEmpty()) {
        // 创建带指数退避策略的请求初始化器
        HttpRequestInitializer retryInitializer = request -> {
            gmailClient(googleId).getRequestInitializer().initialize(request);
            request.setUnsuccessfulResponseHandler(
                    new HttpBackOffUnsuccessfulResponseHandler(
                            new ExponentialBackOff.Builder()
                                    .setInitialIntervalMillis(1000)
                                    .setMaxIntervalMillis(10000)
                                    .build()
                    )
            );
        };

        // 构建用于重试的Gmail客户端
        Gmail retryGmailClient = new Gmail.Builder(
                gmailClient(googleId).getTransport(),
                gmailClient(googleId).getJsonFactory(),
                retryInitializer
        )
                .setApplicationName(gmailClient(googleId).getApplicationName())
                .build();

        // 逐个重试失败的请求
        for (String msgId : failedMessageIds) {
            try {
                Message message = retryGmailClient.users().messages().get(googleId, msgId).setFormat("raw").execute();
                messages.add(MessageMapper.getMessage(googleId, message, false));
            } catch (IOException e) {
                log.error("消息 {} 重试失败: {}", msgId, e.getMessage());
            }
        }
    }
}

2. 拆分批量请求，减小单次批量大小 #

把原来一次批量50个请求拆成更小的批次（比如10个一批），每批执行完后短暂等待，降低请求频率，减少触发限流的概率。

示例代码：

List<MessageDto> messages = new ArrayList<>();
String nextPageToken = null;

List<Message> batchMessages = new ArrayList<>();
ListMessagesResponse response =
        gmailClient(googleId)
                .users()
                .messages()
                .list(googleId)
                .setQ(query)
                .setPageToken(pageToken)
                .setMaxResults(maxResults == null ? 50 : maxResults)
                .execute();

if (response.getMessages() != null) {
    batchMessages.addAll(response.getMessages());
    nextPageToken = response.getNextPageToken();

    // 拆分批次，每10个请求为一批
    int subBatchSize = 10;
    for (int i = 0; i < batchMessages.size(); i += subBatchSize) {
        List<Message> subBatch = batchMessages.subList(i, Math.min(i + subBatchSize, batchMessages.size()));
        
        final JsonBatchCallback<Message> callback =
                new JsonBatchCallback<Message>() {
                    @Override
                    public void onSuccess(Message message, HttpHeaders responseHeaders) {
                        messages.add(MessageMapper.getMessage(googleId, message, false));
                    }

                    @Override
                    public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
                        log.error("处理消息失败: {}", e.getMessage());
                    }
                };
        BatchRequest batch = gmailClient(googleId).batch();

        for (Message message : subBatch) {
            gmailClient(googleId)
                    .users()
                    .messages()
                    .get(googleId, message.getId())
                    .setFormat("raw")
                    .queue(batch, callback);
        }
        batch.execute();
        
        // 每批执行完后等待1秒，可根据实际情况调整等待时长
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

3. 在回调中手动实现退避重试 #

针对429错误，在onFailure回调里手动计算退避时间，设置最大重试次数后重新发起单个请求。

示例回调代码片段：

// 注意：需确保messages、gmailClient、msgId在回调中可访问（可通过final变量或成员变量实现）
@Override
public void onFailure(GoogleJsonError e, HttpHeaders responseHeaders) {
    log.error("处理消息 {} 失败: {}", msgId, e.getMessage());
    if (e.getCode() == 429) {
        int maxRetries = 3;
        long backOffMillis = 1000; // 初始等待1秒
        for (int retryCount = 0; retryCount < maxRetries; retryCount++) {
            try {
                Thread.sleep(backOffMillis);
                Message message = gmailClient(googleId).users().messages().get(googleId, msgId).setFormat("raw").execute();
                messages.add(MessageMapper.getMessage(googleId, message, false));
                return; // 重试成功，退出循环
            } catch (IOException | InterruptedException ex) {
                log.error("消息 {} 第 {} 次重试失败: {}", msgId, retryCount + 1, ex.getMessage());
                backOffMillis *= 2; // 指数退避，每次等待时间翻倍
                if (Thread.currentThread().isInterrupted()) {
                    Thread.currentThread().interrupt();
                    break;
                }
            }
        }
        log.error("消息 {} 所有重试均失败", msgId);
    }
}

内容的提问来源于stack exchange，提问作者Swordfish