如何用Java实现带正则校验的AVRO自定义Logical Type

你已经走完了大半流程，核心问题就是如何从Schema中提取自定义的正则规则，以及如何在LogicalType与Conversion之间传递这个配置。下面一步步帮你完善代码，完全对齐你给出的JS示例场景：

1. 修复自定义LogicalType，从Schema读取正则规则 #

Avro的LogicalType初始化时可以拿到对应的Schema，我们需要在构造方法里从Schema的properties中提取pattern参数。修改你的RegExpLogicalType：

import org.apache.avro.LogicalType;
import org.apache.avro.Schema;
import java.util.regex.Pattern;

public class RegExpLogicalType extends LogicalType {
    public static final String REGEXP_LOGICAL_TYPE_NAME = "regexp";
    private final Pattern pattern;

    // 从Schema中提取pattern参数的构造方法
    public RegExpLogicalType(Schema schema) {
        super(REGEXP_LOGICAL_TYPE_NAME);
        // 读取Schema中定义的pattern属性
        String patternStr = (String) schema.getObjectProp("pattern");
        if (patternStr == null || patternStr.isEmpty()) {
            throw new IllegalArgumentException("'pattern' property is required for regexp logical type");
        }
        this.pattern = Pattern.compile(patternStr);
    }

    // 对外提供获取pattern的方法，供Conversion调用
    public Pattern getPattern() {
        return pattern;
    }

    @Override
    public Schema.LogicalTypeFactory getLogicalTypeFactory() {
        // 实现工厂类，让Avro能根据Schema创建我们的LogicalType实例
        return new Schema.LogicalTypeFactory() {
            @Override
            public LogicalType fromSchema(Schema schema) {
                return new RegExpLogicalType(schema);
            }
        };
    }

    @Override
    public void validate(Schema schema) {
        super.validate(schema);
        // 强制校验：这个LogicalType只能作用在string类型上（和你的JS示例一致）
        if (schema.getType() != Schema.Type.STRING) {
            throw new IllegalArgumentException("Regexp logical type only applies to string schema");
        }
    }
}

这里的核心点：

• 通过schema.getObjectProp("pattern")读取你在Schema中定义的正则规则
• 实现LogicalTypeFactory，让Avro能够自动根据Schema实例化你的自定义类型
• 重写validate方法，确保类型只用于字符串校验，避免误用

2. 修改Conversion类，获取LogicalType中的正则规则 #

现在在RegExpConversion里，我们可以把传入的LogicalType强转为自己的RegExpLogicalType，然后调用getPattern()拿到预定义的正则：

import org.apache.avro.Conversion;
import org.apache.avro.LogicalType;
import org.apache.avro.Schema;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.regex.Matcher;

public class RegExpConversion extends Conversion<String> {
    private static final RegExpConversion INSTANCE = new RegExpConversion();

    public static RegExpConversion get() {
        return INSTANCE;
    }

    private RegExpConversion() {
        super();
    }

    @Override
    public Class<String> getConvertedType() {
        // 我们的逻辑类型对应Java的String类型（和基础类型一致，仅增加校验）
        return String.class;
    }

    @Override
    public String getLogicalTypeName() {
        return RegExpLogicalType.REGEXP_LOGICAL_TYPE_NAME;
    }

    @Override
    public String fromBytes(ByteBuffer value, Schema schema, LogicalType type) {
        // 强转为自定义LogicalType，获取预定义的正则
        RegExpLogicalType regexpType = (RegExpLogicalType) type;
        String strValue = StandardCharsets.UTF_8.decode(value).toString();
        
        // 执行正则校验
        Matcher matcher = regexpType.getPattern().matcher(strValue);
        if (!matcher.matches()) {
            throw new IllegalArgumentException(
                String.format("Value '%s' does not match pattern '%s'", strValue, regexpType.getPattern().pattern())
            );
        }
        return strValue;
    }

    @Override
    public ByteBuffer toBytes(String value, Schema schema, LogicalType type) {
        // 序列化前同样做校验
        RegExpLogicalType regexpType = (RegExpLogicalType) type;
        Matcher matcher = regexpType.getPattern().matcher(value);
        if (!matcher.matches()) {
            throw new IllegalArgumentException(
                String.format("Value '%s' does not match pattern '%s'", value, regexpType.getPattern().pattern())
            );
        }
        return ByteBuffer.wrap(value.getBytes(StandardCharsets.UTF_8));
    }

    // 支持JSON反序列化时的直接字符串校验
    @Override
    public String fromString(String value, Schema schema, LogicalType type) {
        RegExpLogicalType regexpType = (RegExpLogicalType) type;
        Matcher matcher = regexpType.getPattern().matcher(value);
        if (!matcher.matches()) {
            throw new IllegalArgumentException(
                String.format("Value '%s' does not match pattern '%s'", value, regexpType.getPattern().pattern())
            );
        }
        return value;
    }
}

这里调整了Conversion的泛型为String，更符合语义（我们是对字符串做校验），同时在序列化和反序列化的关键节点都加入了校验逻辑。

3. 注册自定义LogicalType和Conversion #

要让Avro识别你的自定义类型，需要在应用启动时完成注册：

import org.apache.avro.LogicalTypes;
import org.apache.avro.data.Conversions;

public class AvroCustomTypeSetup {
    public static void register() {
        // 注册LogicalType工厂
        LogicalTypes.register(RegExpLogicalType.REGEXP_LOGICAL_TYPE_NAME, 
                              new RegExpLogicalType(null).getLogicalTypeFactory());
        // 注册Conversion
        Conversions.add(RegExpConversion.get());
    }
}

4. 完整使用示例（对齐你的JS场景） #

先定义Avro Schema（JSON格式，命名为example.avsc）：

{
  "type": "record",
  "name": "Example",
  "fields": [
    {
      "name": "custId",
      "type": "string"
    },
    {
      "name": "sessionId",
      "type": {
        "type": "string",
        "logicalType": "regexp",
        "pattern": "^\\d{3}-\\d{4}-\\d{5}$"
      }
    }
  ]
}

然后在Java中使用：

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Paths;

public class AvroValidationDemo {
    public static void main(String[] args) throws IOException {
        // 先注册自定义类型
        AvroCustomTypeSetup.register();

        // 加载Schema文件
        String schemaJson = new String(Files.readAllBytes(Paths.get("example.avsc")));
        Schema schema = new Schema.Parser().parse(schemaJson);

        // 测试合法数据
        GenericRecord validRecord = new GenericData.Record(schema);
        validRecord.put("custId", "abc");
        validRecord.put("sessionId", "123-1234-12345");
        try {
            byte[] bytes = GenericData.get().toBytes(validRecord);
            System.out.println("合法数据序列化成功");
        } catch (Exception e) {
            e.printStackTrace();
        }

        // 测试非法数据
        GenericRecord invalidRecord = new GenericData.Record(schema);
        invalidRecord.put("custId", "abc");
        invalidRecord.put("sessionId", "foobar");
        try {
            byte[] bytes = GenericData.get().toBytes(invalidRecord);
            System.out.println("非法数据未触发校验（异常）");
        } catch (IllegalArgumentException e) {
            System.out.println("触发预期校验异常：" + e.getMessage());
            // 输出：触发预期校验异常：Value 'foobar' does not match pattern '^\d{3}-\d{4}-\d{5}$'
        }
    }
}

核心逻辑总结 #

• Avro的自定义LogicalType通过Schema的properties传递配置参数，所以你在Schema中写的pattern会存在schema.getObjectProp("pattern")中
• 必须实现LogicalTypeFactory，否则Avro无法根据Schema自动创建你的自定义LogicalType实例
• Conversion方法中通过强转LogicalType获取预定义的正则规则，从而在序列化/反序列化过程中自动完成校验

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