AWS Marketplace BatchMeterUsage Java API集成咨询：批量计量数据上报

核心改造思路 #

• 替换原单条调用MeterUsage的逻辑，改用BatchMeterUsage接口批量提交（单次最多25条）
• 新增RestAPI接口接收待上报计量数据，同时保留定时批量触发逻辑（支持手动触发）
• 用本地缓存/队列暂存计量数据，达到25条或整点时触发批量上报

具体实现步骤 #

1. 依赖准备 #

确保AWS SDK依赖已引入（以Maven为例）：

<dependency>
    <groupId>software.amazon.awssdk</groupId>
    <artifactId>marketplacemetering</artifactId>
    <version>2.20.0</version>
</dependency>

2. 批量计量服务改造 #

在已有服务类中新增批量上报方法，封装BatchMeterUsage调用：

import software.amazon.awssdk.services.marketplacemetering.MarketplaceMeteringClient;
import software.amazon.awssdk.services.marketplacemetering.model.*;
import java.util.List;

@Service
public class MeteringService {

    private final MarketplaceMeteringClient meteringClient;

    // 构造注入AWS客户端
    public MeteringService(MarketplaceMeteringClient meteringClient) {
        this.meteringClient = meteringClient;
    }

    // 批量上报计量数据
    public BatchMeterUsageResponse batchReportMeterData(List<UsageRecord> usageRecords) {
        // 校验单次批量数量不超过25条
        if (usageRecords.size() > 25) {
            throw new IllegalArgumentException("单次批量上报最多支持25条记录");
        }

        BatchMeterUsageRequest request = BatchMeterUsageRequest.builder()
                .productCode("你的AWS Marketplace产品编码")
                .usageRecords(usageRecords)
                .build();

        return meteringClient.batchMeterUsage(request);
    }
}

3. RestAPI控制器实现 #

新增接口接收计量数据，同时加入批量触发上报接口：

import org.springframework.web.bind.annotation.*;
import software.amazon.awssdk.services.marketplacemetering.model.UsageRecord;
import java.util.ArrayList;
import java.util.List;

@RestController
@RequestMapping("/api/metering")
public class MeteringController {

    private final MeteringService meteringService;
    // 本地缓存待上报的计量数据（生产环境建议用Redis/消息队列替代）
    private final List<UsageRecord> pendingRecords = new ArrayList<>();

    public MeteringController(MeteringService meteringService) {
        this.meteringService = meteringService;
    }

    // 接收单条计量数据，自动累积到批量队列
    @PostMapping("/record")
    public void submitMeterRecord(@RequestBody UsageRecord record) {
        synchronized (pendingRecords) {
            pendingRecords.add(record);
            // 达到25条时立即触发批量上报
            if (pendingRecords.size() >= 25) {
                List<UsageRecord> batch = new ArrayList<>(pendingRecords);
                pendingRecords.clear();
                meteringService.batchReportMeterData(batch);
            }
        }
    }

    // 手动触发批量上报（配合定时任务使用）
    @PostMapping("/batch-report")
    public void triggerBatchReport() {
        synchronized (pendingRecords) {
            if (!pendingRecords.isEmpty()) {
                List<UsageRecord> batch = new ArrayList<>(pendingRecords);
                pendingRecords.clear();
                meteringService.batchReportMeterData(batch);
            }
        }
    }
}

4. 定时任务改造 #

替换原TaskTimer的单条上报逻辑，改为调用批量上报接口：

import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Component
public class MeteringTaskTimer {

    private final MeteringController meteringController;

    public MeteringTaskTimer(MeteringController meteringController) {
        this.meteringController = meteringController;
    }

    // 每小时整点触发批量上报，确保未达25条的数据也能及时上报
    @Scheduled(cron = "0 0 * * * ?")
    public void hourlyBatchMetering() {
        meteringController.triggerBatchReport();
    }
}

5. 异常处理与重试机制 #

在批量上报方法中加入重试逻辑，处理AWS服务异常：

import software.amazon.awssdk.core.exception.SdkException;
import org.springframework.retry.annotation.Retryable;

@Service
public class MeteringService {
    // ... 原有代码

    @Retryable(value = {SdkException.class}, maxAttempts = 3)
    public BatchMeterUsageResponse batchReportMeterData(List<UsageRecord> usageRecords) {
        // ... 原有上报逻辑
    }
}

需引入Spring Retry依赖：

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
    <version>1.3.4</version>
</dependency>
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-aop</artifactId>
</dependency>

注意事项 #

• 生产环境不要用内存列表缓存数据，建议用Redis或消息队列（如SQS）保证数据不丢失
• 需对UsageRecord的customerIdentifier、dimension、quantity等字段做合法性校验
• 要处理BatchMeterUsageResponse中的失败记录，对失败条目进行重试或存入死信队列

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