Apache NiFi-自定义控制器服务

简介

Apache NiFi是一个基于流程编排的开源数据集成系统，使得数据可操作、管理和监控。它支持高效、可扩展、可编程的数据流传输、缓存和管理。自定义控制器服务是NiFi中的一个功能，可以通过它来自定义数据流中所需的资源。

什么是自定义控制器服务？

NiFi中的控制器服务是一个共享的全局资源，可以被多个处于同一NiFi实例的组件进行访问。它们提供了NiFi核心机制之外的通用功能，并可以大幅简化数据流的构建。自定义控制器服务就是用户自己开发的控制器服务，NiFi中没有提供该服务。

通过开发自定义控制器服务，可以将自己开发的功能作为NiFi的服务提供出去，其他数据流组件就可以使用该服务，极大地方便了数据流构建。

如何开发自定义控制器服务？

开发自定义控制器服务非常简单，你只需要实现ControllerService接口即可，ControllerService中定义了一些方法，你需要实现这些方法来提供服务：

public interface ControllerService {
    /**
     * @return the Fully Qualified Class Name of this Controller service.
     */
    String getIdentifier();

    /**
     * Will be called when the service is first initialized.
     * @param context configuration context
     * @throws InitializationException if unable to init
     */
    void init(ControllerServiceInitializationContext context) throws InitializationException;

    /**
     * @return the list of PropertyDescriptors for this service
    */
    List<PropertyDescriptor> getSupportedPropertyDescriptors();

    /**
     * 开启服务
    */
    void onEnabled();

    /**
     * 关闭服务
    */
    void onDisabled();

    /**
     * 若服务需要在调用时打开某些资源，例如数据库连接，则需要再`onEnabled`方法中打开资源，在`onDisabled`方法中关闭资源，这样才能正确地实现生命周期管理。
    */

    /**
     * Validates the given configuration.
     *
     * @param context configuration context
     */
    Collection<ValidationResult> validate(ControllerServiceInitializationContext context);

    /**
     * Will be invoked when the component which owns this Controller Service instance invokes start of the owning component.
     */
    void start();

    /**
     * Will be invoked when the component which owns this Controller Service instance invokes stop of the owning component.
     */
    void stop();
}

除此之外，在开发自定义控制器服务时，还需要注意以下几点：

1. 实现getSupportedPropertyDescriptors方法来提供你的服务允许使用的属性列表。
2. 在onEnabled方法中打开资源，在onDisabled方法中关闭资源，以实现正确的服务生命周期管理。
3. 提供服务的具体实现。

一般来说，你还需要编写一个ControllerService实现类，用来继承AbstractControllerService类，并实现必要的方法，其中最重要的就是服务提供的具体实现。

举一个简单的例子，假设我们要开发一个将文本转为小写的服务。首先，我们需要实现ControllerService接口。

public interface LowerCaseService extends ControllerService {
    String toLowerCase(String text);
}

然后，我们需要实现toLowerCase方法，提供具体的服务功能。

public class LowerCaseServiceImpl extends AbstractControllerService implements LowerCaseService {
    @Override
    public String toLowerCase(String text) {
        return text.toLowerCase();
    }
}

最后，我们需要提供服务的属性和服务的实现，即getSupportedPropertyDescriptors方法和onEnabled方法。

public class LowerCaseControllerService extends AbstractControllerService implements LowerCaseService {
    private volatile int maxStringLength = 1000;

    public static final PropertyDescriptor MAX_STRING_LENGTH = new PropertyDescriptor.Builder()
        .name("Max String Length")
        .description("The maximum length of a string that can be converted to lowercase.")
        .defaultValue("1000")
        .addValidator(StandardValidators.POSITIVE_INTEGER_VALIDATOR)
        .required(true)
        .build();

    @Override
    public String toLowerCase(String text) {
        int length = text.length();
        if (length > maxStringLength) {
            throw new IllegalArgumentException("Text is too long for this service.");
        }

        return text.toLowerCase();
    }

    @Override
    public List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        final List<PropertyDescriptor> properties = new ArrayList<>();
        properties.add(MAX_STRING_LENGTH);
        return properties;
    }

    @OnEnabled
    public void onEnabled() {
        maxStringLength = getContext().getProperty(MAX_STRING_LENGTH).asInteger();
    }
}

总结

自定义控制器服务是NiFi中非常重要的一部分，它为NiFi提供了扩展性和可定制性。自定义控制器服务的开发非常简单，只需要实现ControllerService即可。通过自定义控制器服务，你可以将自己开发的功能作为服务提供出去，其他数据流组件可以方便地使用。