如何在Apache Flink中注册java.util.List类型

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-java</artifactId>
    <version>1.13.2</version>
</dependency>

import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.TypeExtractor;
import org.apache.flink.api.java.typeutils.TypeInformationFactory;
import org.apache.flink.api.java.typeutils.TypeInformationResult;

import java.util.List;

public class ListTypeInfo<T> extends TypeInformation<List<T>> {

    private final TypeInformation<T> elementType;

    public ListTypeInfo(TypeInformation<T> elementType) {
        this.elementType = elementType;
    }

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

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

    @Override
    public int getArity() {
        return 1;
    }

    @Override
    public int getTotalFields() {
        return 1;
    }

    @Override
    public Class<List<T>> getTypeClass() {
        return (Class<List<T>>) (Class<?>) List.class;
    }

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

    @Override
    public TypeSerializer<List<T>> createSerializer(ExecutionConfig config) {
        return new ListSerializer<>(elementType.createSerializer(config));
    }

    @Override
    public String toString() {
        return "List<" + elementType + ">";
    }

    public static <T> TypeInformation<List<T>> of(TypeInformation<T> elementType) {
        return new ListTypeInfo<>(elementType);
    }

    public static <T> TypeInformation<List<T>> of(Class<T> elementType) {
        return new ListTypeInfo<>(TypeExtractor.getForClass(elementType));
    }

    public static <T> TypeInformationResult<List<T>> createTypeInfo(TypeInformationFactory<List<T>> factory, TypeInformation<?>[] parameters, Annotation[] annotations) {
        return TypeInformationResult.newBuilder(factory.createTypeInfo(parameters[0], annotations)).build();
    }
}

import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.typeutils.TypeExtractor;

public class FlinkListRegistrationExample {

    public static void main(String[] args) throws Exception {
        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
        env.getConfig().registerTypeWithKryoSerializer(List.class, new ListTypeInfo<>(TypeExtractor.getForClass(Object.class)));
        
        // 其他Flink程序逻辑
    }
}