- **概述**
在使用 Kafka 的 SimpleStringSchema 时，如果在调用其方法时传入了一个 null 值，则可能会导致空指针异常。
为了避免这种情况，您应该在使用 SimpleStringSchema 的方法之前检查传入的值是否为 null，并在必要时使用一个非 null 值代替。
我们看下为何会造成空指针：
```java
    public String deserialize(byte[] message) {
        //message可能为null
        return new String(message, this.charset);
    }
```

- **推荐方案一**
使用自定义的：
```java
        return new FlinkKafkaConsumer<String>(topic,
                new KafkaDeserializationSchema<String>() {
                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

@Override
                    public String deserialize(ConsumerRecord<byte[], byte[]> record) throws Exception {
                        if (record == null || record.value() == null) {
                            return "";
                        } else {
                            return new String(record.value());
                        }
                    }

@Override
                    public TypeInformation<String> getProducedType() {
                        return BasicTypeInfo.STRING_TYPE_INFO;
                    }
                },
                properties);
```

- **推荐方案二**
重写一个SimpleStringSchema类
```java
/**
 * Very simple serialization schema for strings.
 *
 * <p>By default, the serializer uses "UTF-8" for string/byte conversion.
 */
@PublicEvolving
public class MyStringSchema
        implements DeserializationSchema<String>, SerializationSchema<String> {

private static final long serialVersionUID = 1L;

/**
     * The charset to use to convert between strings and bytes. The field is transient because we
     * serialize a different delegate object instead
     */
    private transient Charset charset;

/** Creates a new SimpleStringSchema that uses "UTF-8" as the encoding. */
    public MyStringSchema() {
        this(StandardCharsets.UTF_8);
    }

/**
     * Creates a new SimpleStringSchema that uses the given charset to convert between strings and
     * bytes.
     *
     * @param charset The charset to use to convert between strings and bytes.
     */
    public MyStringSchema(Charset charset) {
        this.charset = checkNotNull(charset);
    }

/**
     * Gets the charset used by this schema for serialization.
     *
     * @return The charset used by this schema for serialization.
     */
    public Charset getCharset() {
        return charset;
    }

// ------------------------------------------------------------------------
    //  Kafka Serialization
    // ------------------------------------------------------------------------

@Override
    public String deserialize(byte[] message) {
        if(Objects.isNull(message)){
            return "";
        }
        return new String(message, charset);
    }

@Override
    public boolean isEndOfStream(String nextElement) {
        return false;
    }

@Override
    public byte[] serialize(String element) {
        return element.getBytes(charset);
    }

@Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }

// ------------------------------------------------------------------------
    //  Java Serialization
    // ------------------------------------------------------------------------

private void writeObject(ObjectOutputStream out) throws IOException {
        out.defaultWriteObject();
        out.writeUTF(charset.name());
    }

private void readObject(java.io.ObjectInputStream in)
            throws IOException, ClassNotFoundException {
        in.defaultReadObject();
        String charsetName = in.readUTF();
        this.charset = Charset.forName(charsetName);
    }
}
```

ITPub博客

Kafka SimpleStringSchema 可能会造成空指针异常

最新文章