响应式Java中RabbitMQ消息反序列化失败问题求助

先排查核心错误根源 #

错误invalid stream header: 4D657373对应的ASCII是"Mess"，而Java对象序列化的字节流开头固定是AC ED 00 05，这说明你收到的body根本不是Java序列化生成的字节数据。优先检查发送端：

• 确认发送端是通过ObjectOutputStream将CustomModel序列化为byte[]后，再放入OutboundMessage的body字段
• 排查发送端是否有额外字符串前缀（比如"Message:"这类标识）被写入body，导致字节流开头不符合Java序列化规范

修正Reactor中的反序列化逻辑 #

即使发送端修复后，也要注意Reactor的线程模型：ObjectInputStream操作属于阻塞IO，不能在Reactor默认的非阻塞线程池中执行，否则会阻塞线程池影响整体性能。需要将反序列化操作切换到阻塞IO专用调度器：

调整Flux的处理流程 #

将原来的map配合publishOn切换线程，或者用Mono.fromCallable包装阻塞操作：

Flux<CustomModel> messages = receiver.consumeNoAck("queue")
        // 将阻塞反序列化操作切换到boundedElastic调度器
        .publishOn(Schedulers.boundedElastic())
        .map(this::deliveryToCustomModel)
        .doOnNext(customModel -> {
            // 执行业务逻辑
        })
        .doOnError(e -> {
            // 处理反序列化错误，比如记录日志
            System.err.println("反序列化失败: " + e.getMessage());
        })
        .subscribe();

优化反序列化方法 #

不要吞掉异常（原代码catch后仅打印消息并返回null），让异常向上抛出，便于Reactor的错误机制处理：

private CustomModel deliveryToCustomModel(Delivery delivery) throws IOException, ClassNotFoundException {
    try (ObjectInputStream inputStream = new ObjectInputStream(new ByteArrayInputStream(delivery.getBody()))) {
        return (CustomModel) inputStream.readObject();
    }
}

如果需要跳过反序列化失败的消息，可结合filter处理：

Flux<CustomModel> messages = receiver.consumeNoAck("queue")
        .publishOn(Schedulers.boundedElastic())
        .map(delivery -> {
            try {
                return deliveryToCustomModel(delivery);
            } catch (Exception e) {
                System.err.println("跳过错误消息: " + e.getMessage());
                return null;
            }
        })
        .filter(Objects::nonNull) // 过滤反序列化失败的null对象
        .doOnNext(customModel -> {
            // 执行业务逻辑
        })
        .subscribe();

额外注意事项 #

• 确保CustomModel实现了Serializable接口，且发送端与接收端的类结构完全一致（包括包名、类名、字段定义），否则会抛出ClassNotFoundException
• 若发送端与接收端的CustomModel存在版本差异，需添加serialVersionUID来保证序列化兼容性

内容的提问来源于stack exchange，提问作者Алишер Асхат