异步接口调用说明

4.3.1 配置

以下是配置项说明，不要把真实密码或密钥提交到仓库。示例从环境变量读取，生产环境可改为密钥管理系统或受控配置注入。

ASYNC_MQ_ENDPOINTS=rmq-cn-x684z19du01.cn-beijing.rmq.aliyuncs.com:8080
ASYNC_MQ_NAMESPACE=rmq-cn-x684z19du01
ASYNC_MQ_TOPIC=ASYNC_TASK_RESULT_TOPIC
ASYNC_MQ_GROUP=CID_ASYNC_TASK_RESULT_{UID}
ASYNC_MQ_USERNAME=<消费账号：询问服务提供方>
ASYNC_MQ_PASSWORD=<消费密码：询问服务提供方>
ASYNC_MQ_UID=1234567890
ASYNC_MQ_API_TYPES=api-a,api-b
ASYNC_RESULT_KEY_SECRET=<该UID的用户密钥：询问服务提供方>

4.3.2 消费、解密与确认

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.StaticSessionCredentialsProvider;
import org.apache.rocketmq.client.apis.consumer.ConsumeResult;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.FilterExpressionType;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import java.nio.ByteBuffer;
import java.util.Arrays;
import java.util.Base64;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.logging.Logger;
import java.util.stream.Collectors;

public final class AsyncTaskResultConsumer {
    private static final Logger LOG = Logger.getLogger(AsyncTaskResultConsumer.class.getName());

    @FunctionalInterface
    public interface ResultHandler {
        // 必须在业务数据可靠持久化或幂等确认已处理后才正常返回；失败抛异常。
        void handle(String taskId, String tag, JSONObject result) throws Exception;
    }

    public static PushConsumer start(ResultHandler handler) throws Exception {
        Objects.requireNonNull(handler, "handler");
        String uid = required("ASYNC_MQ_UID");
        Set<String> tags = Arrays.stream(required("ASYNC_MQ_API_TYPES").split(",", -1))
                .map(String::trim)
                .map(apiType -> {
                    if (apiType.isEmpty() || apiType.contains("|") || apiType.contains("*")) {
                        throw new IllegalArgumentException("Invalid apiType");
                    }
                    return uid + "-" + apiType;
                })
                .collect(Collectors.toSet());
        String keySecret = required("ASYNC_RESULT_KEY_SECRET");
        if (Base64.getDecoder().decode(keySecret).length != 32) {
            throw new IllegalArgumentException("Expected a 32-byte AES key");
        }
        var config = ClientConfiguration.newBuilder()
                .setEndpoints(required("ASYNC_MQ_ENDPOINTS"))
                .setCredentialProvider(new StaticSessionCredentialsProvider(
                        required("ASYNC_MQ_USERNAME"), required("ASYNC_MQ_PASSWORD")));
        String namespace = System.getenv("ASYNC_MQ_NAMESPACE");
        if (namespace != null && !namespace.isBlank()) {
            config.setNamespace(namespace);
        }
        FilterExpression filter = new FilterExpression(
                tags.stream().sorted().collect(Collectors.joining("||")),
                FilterExpressionType.TAG);

        return ClientServiceProvider.loadService().newPushConsumerBuilder()
                .setClientConfiguration(config.build())
                .setConsumerGroup(required("ASYNC_MQ_GROUP"))
                .setSubscriptionExpressions(Map.of(required("ASYNC_MQ_TOPIC"), filter))
                .setConsumptionThreadCount(8)
                .setMessageListener(message -> {
                    try {
                        String tag = message.getTag().orElseThrow();
                        if (!tags.contains(tag)) {
                            throw new IllegalArgumentException("Unexpected message tag");
                        }
                        ByteBuffer buffer = message.getBody().duplicate();
                        byte[] body = new byte[buffer.remaining()];
                        buffer.get(body); // 不假定 ByteBuffer 一定有可访问的 array。
                        JSONObject result = AsyncTaskResultDecryptor.decryptMessage(body, keySecret);
                        JSONObject data = result.getJSONObject("Data");
                        if (data == null || data.getString("TaskId") == null
                                || data.getString("TaskId").isBlank()) {
                            throw new IllegalArgumentException("Data.TaskId is required");
                        }
                        String status = data.getString("Status");
                        if (status == null || !Set.of("COMPLETED", "FAILED", "CANCELLED").contains(status)) {
                            throw new IllegalArgumentException("Unexpected task status");
                        }
                        handler.handle(data.getString("TaskId"), tag, result);
                        return ConsumeResult.SUCCESS;
                    } catch (Exception e) {
                        // 不记录凭据、消息体、明文或异常中的业务内容。
                        LOG.warning("Async result consumption failed, messageId="
                                + message.getMessageId() + ", errorType=" + e.getClass().getSimpleName());
                        return ConsumeResult.FAILURE;
                    }
                })
                .build();
    }

    private static String required(String name) {
        String value = System.getenv(name);
        if (value == null || value.isBlank()) {
            throw new IllegalArgumentException("Missing environment variable: " + name);
        }
        return value;
    }
}