如何在Flink (Java)中序列化Tuple3？

import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.core.memory.DataInputView;
import org.apache.flink.core.memory.DataOutputView;
import org.apache.flink.types.CopyableValue;

import java.io.IOException;

public class Tuple3Serializer<T1, T2, T3> implements TypeSerializer<Tuple3<T1, T2, T3>> {

    @Override
    public void serialize(Tuple3<T1, T2, T3> tuple, DataOutputView dataOutputView) throws IOException {
        // 将Tuple3对象的字段按照需要的格式写入DataOutputView
        dataOutputView.writeUTF(tuple.f0.toString());
        dataOutputView.writeUTF(tuple.f1.toString());
        dataOutputView.writeUTF(tuple.f2.toString());
    }

    @Override
    public Tuple3<T1, T2, T3> deserialize(DataInputView dataInputView) throws IOException {
        // 从DataInputView中读取字段，并创建一个新的Tuple3对象
        T1 field1 = (T1) dataInputView.readUTF();
        T2 field2 = (T2) dataInputView.readUTF();
        T3 field3 = (T3) dataInputView.readUTF();
        return new Tuple3<>(field1, field2, field3);
    }

    @Override
    public Tuple3<T1, T2, T3> deserialize(Tuple3<T1, T2, T3> reuse, DataInputView dataInputView) throws IOException {
        // 从DataInputView中读取字段，并更新已存在的Tuple3对象
        reuse.f0 = (T1) dataInputView.readUTF();
        reuse.f1 = (T2) dataInputView.readUTF();
        reuse.f2 = (T3) dataInputView.readUTF();
        return reuse;
    }

    @Override
    public void copy(DataInputView dataInputView, DataOutputView dataOutputView) throws IOException {
        // 复制字节流
        byte[] buffer = new byte[1024];
        int bytesRead;
        while ((bytesRead = dataInputView.read(buffer)) != -1) {
            dataOutputView.write(buffer, 0, bytesRead);
        }
    }

    @Override
    public boolean isImmutableType() {
        return false;
    }

    @Override
    public TypeSerializer<Tuple3<T1, T2, T3>> duplicate() {
        return this;
    }

    @Override
    public int getLength() {
        return -1;
    }

    @Override
    public void serialize(Tuple3<T1, T2, T3> record, DataOutputView target) throws IOException {
        serialize(record, target);
    }

    @Override
    public Tuple3<T1, T2, T3> deserialize(DataInputView source) throws IOException {
        return deserialize(source);
    }

    @Override
    public Tuple3<T1, T2, T3> deserialize(Tuple3<T1, T2, T3> reuse, DataInputView source) throws IOException {
        return deserialize(reuse, source);
    }

    @Override
    public void copy(DataInputView source, DataOutputView target) throws IOException {
        copy(source, target);
    }

    @Override
    public boolean equals(Object obj) {
        return obj instanceof Tuple3Serializer;
    }

    @Override
    public int hashCode() {
        return getClass().hashCode();
    }
}

import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class FlinkSerializationExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 创建自定义的序列化器
        Tuple3Serializer<String, Integer, Double> tuple3Serializer = new Tuple3Serializer<>();

        // 注册自定义的序列化器
        ExecutionConfig config = env.getConfig();
        config.registerTypeWithKryoSerializer(Tuple3.class, tuple3Serializer);

        // 使用自定义的序列化器进行序列化和反序列化操作
        Tuple3<String, Integer, Double> tuple = new Tuple3<>("value1", 2, 3.14);
        byte[] serializedTuple = env.getSerializer(Tuple3.class).serialize(tuple);
        Tuple3<String, Integer, Double> deserializedTuple = env.getSerializer(Tuple3.class).deserialize(serializedTuple);

        System.out.println("Original Tuple: " + tuple);
        System.out.println("Serialized Tuple: " + serializedTuple);
        System.out.println("Deserialized Tuple: " + deserializedTuple);

        env.execute("Flink Serialization Example");
    }
}