Micronaut调用API上传字节数据：监听完成事件及Flux订阅错误解决

问题1：监听上传完成事件 #

Micronaut HTTP Client 基于响应式编程模型，返回的Publisher需要通过正确的订阅回调来监听上传完成（成功/失败）事件。你可以利用Reactor的Flux/Mono提供的多参数subscribe方法，分别处理成功响应、错误和完成回调：

修正后的上传代码示例 #

// 构建Multipart请求
MultipartBody requestBody = MultipartBody.builder()
        .addPart("data", uploadFile.getFilename(), fileContentBytes)
        .build();

// 创建HttpClient（建议复用，不要每次请求新建）
HttpClient httpClient = HttpClient.create(new URL(uploadUrl.getUpload_url()));

HttpRequest<?> uploadRequest = HttpRequest.POST(uploadUrl.getUpload_url(), requestBody)
        .contentType(MediaType.MULTIPART_FORM_DATA_TYPE)
        .header(HttpHeaders.AUTHORIZATION, v2Request.getToken());

// 转换为Flux并订阅，监听完成事件
Flux.from(httpClient.exchange(uploadRequest, HttpResponse.class))
        .subscribe(
                // 成功拿到响应时的处理
                response -> {
                    if (response.status().isSuccess()) {
                        // 上传成功后的逻辑，比如记录日志、更新状态
                        log.info("上传完成，响应状态: {}", response.status());
                    } else {
                        log.error("上传失败，响应状态: {}", response.status());
                    }
                },
                // 发生错误时的处理
                error -> {
                    log.error("上传请求抛出异常", error);
                    // 错误处理逻辑，比如重试、告警
                },
                // 上传完成（请求生命周期结束）的回调
                () -> {
                    log.info("上传请求生命周期结束");
                    // 清理操作，比如关闭临时创建的HttpClient
                    httpClient.close();
                }
        );

注意：HttpClient建议全局复用，不要每次请求都新建，避免资源浪费。如果是Micronaut容器环境，最好通过依赖注入获取HttpClient实例。

问题2：日志输出FluxLift的错误 #

你看到的Response of facebook page post FluxLift是因为错误地打印了Publisher对象本身（response变量是Flux实例），而非实际的HTTP响应内容。Reactor的Flux/Mono对象的toString()会输出内部类名称，而非响应数据。

修正后的日志代码示例 #

HttpClient httpClient = HttpClient.create(new URL(url));
HttpRequest<FbPagePostRequest> request = HttpRequest.POST(url, postRequest)
        .contentType(MediaType.APPLICATION_JSON)
        .header(HttpHeaders.AUTHORIZATION, token);

Publisher<HttpResponse<FbPagePostResponse>> responsePublisher = httpClient.exchange(request, FbPagePostResponse.class);

Flux.from(responsePublisher)
        .map(data -> {
            if (data.status() != HttpStatus.OK || data.getBody().isEmpty()) {
                // 打印实际的响应信息，而非Publisher对象
                log.error("Facebook页面发帖失败，响应状态: {}, 响应体: {}", data.status(), data.getBody());
                throw new PublicationException("Error posting on Facebook on Page : " + pageName);
            }
            return data.getBody().get();
        })
        .subscribe(
                returnObj::set,
                error -> log.error("发帖请求抛出异常", error)
        );

关键修正点 #

• 将log.error("Response of facebook page post " + response)改为打印data（实际的HttpResponse对象）的具体属性，比如状态码、响应体。
• 避免直接打印Publisher类型变量，它们是响应式流的容器，而非实际业务数据。

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