Apache Flink如何将Java ObjectNode转换为JSON字符串输出？

你提到的JSONSerializationSchema确实不是Flink内置的类，不过我们有几种简单的方案可以实现把ObjectNode转换成JSON字符串并通过RMQSink输出的需求，下面给你详细说明：

方案一：先将ObjectNode转为String流，再用SimpleStringSchema #

这是最直接的方式，先通过map操作把ObjectNode转成JSON字符串，再搭配Flink自带的SimpleStringSchema来做序列化：

// 先把ObjectNode转换成JSON字符串
DataStream<String> jsonStringStream = stream.map(new MapFunction<ObjectNode, String>() {
    @Override
    public String map(ObjectNode value) throws Exception {
        // ObjectNode的toString()方法会直接输出标准JSON格式的字符串
        return value.toString();
    }
});

// 将字符串流输出到RabbitMQ
jsonStringStream.addSink(new RMQSink<>(
    connectionConfig,
    "stop",
    new SimpleStringSchema()
));

方案二：自定义针对ObjectNode的SerializationSchema #

如果你不想额外做map转换，可以自己实现一个SerializationSchema，专门处理ObjectNode的序列化：

基础实现（直接用toString） #

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import java.nio.charset.StandardCharsets;

public class ObjectNodeJsonSerializationSchema implements SerializationSchema<ObjectNode> {
    @Override
    public byte[] serialize(ObjectNode element) {
        // 转成UTF-8编码的字节数组
        return element.toString().getBytes(StandardCharsets.UTF_8);
    }
}

然后直接在RMQSink中使用这个自定义Schema：

stream.addSink(new RMQSink<>(
    connectionConfig,
    "stop",
    new ObjectNodeJsonSerializationSchema()
));

进阶实现（用Jackson ObjectMapper） #

如果需要更灵活的序列化配置（比如格式化输出、处理特殊字段），可以用Jackson的ObjectMapper来实现：

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;
import java.io.IOException;

public class ObjectNodeJsonSerializationSchema implements SerializationSchema<ObjectNode> {
    // 建议把ObjectMapper声明为静态常量，避免重复创建
    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

    @Override
    public byte[] serialize(ObjectNode element) throws IOException {
        return OBJECT_MAPPER.writeValueAsBytes(element);
    }
}

这样实现的好处是可以对OBJECT_MAPPER添加自定义配置，比如：

// 示例：开启格式化输出
OBJECT_MAPPER.enable(SerializationFeature.INDENT_OUTPUT);

两种方案都能满足你的需求，你可以根据自己的场景选择～

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