SpringCloud DataFlow — 0. 本地部署部署
SpringCloud DataFlow — 1. 自定义Processor
SpringCloud DataFlow — 2. 自定义Sink
SpringCloud DataFlow — 3. 暴露properties
SpringCloud DataFlow — 4. Prometheus + Grafana 监控
SpringCloud DataFlow — 5. 多分支负载

1. 创建properties类

package etl.dmt.quick.unpackprocessorkafka.config;

import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.validation.annotation.Validated;

import javax.validation.constraints.NotEmpty;
import javax.validation.constraints.NotNull;

/**
 * @author tianshl
 * @version 2019/8/20 3:34 PM
 */
@Validated
@ConfigurationProperties("unpack")
public class UnpackProperties {

    private Integer corePoolSize = 10;
    private Integer keepAliveSeconds = 3;
    private Integer maxPoolSize = 10;
    private Integer queueCapacity = 10000;
    private String rejectedExecutionHandler = "java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy";

    @NotNull
    public Integer getCorePoolSize() {
        return corePoolSize;
    }

    public void setCorePoolSize(Integer corePoolSize) {
        this.corePoolSize = corePoolSize;
    }

    @NotNull
    public Integer getKeepAliveSeconds() {
        return keepAliveSeconds;
    }

    public void setKeepAliveSeconds(Integer keepAliveSeconds) {
        this.keepAliveSeconds = keepAliveSeconds;
    }

    @NotNull
    public Integer getMaxPoolSize() {
        return maxPoolSize;
    }

    public void setMaxPoolSize(Integer maxPoolSize) {
        this.maxPoolSize = maxPoolSize;
    }

    @NotNull
    public Integer getQueueCapacity() {
        return queueCapacity;
    }

    public void setQueueCapacity(Integer queueCapacity) {
        this.queueCapacity = queueCapacity;
    }

    @NotNull
    public String getRejectedExecutionHandler() {
        return rejectedExecutionHandler;
    }

    public void setRejectedExecutionHandler(String rejectedExecutionHandler) {
        this.rejectedExecutionHandler = rejectedExecutionHandler;
    }
}

2. resources目录下新建META-INF文件夹

3. META-INF下新建 spring-configuration-metadata.json

{
  "groups": [
    {
      "name": "clean",
      "type": "etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties",
      "sourceType": "etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties"
    }
  ],
  "properties": [
    {
      "name": "clean.corePoolSize",
      "type": "java.lang.Integer",
      "description": "核心线程数，包括空闲线程",
      "sourceType": "etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties",
      "defaultValue": 10
    },
    {
      "name": "clean.keepAliveSeconds",
      "type": "java.lang.Integer",
      "description": "线程池维护线程所允许的空闲时间",
      "sourceType": "etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties",
      "defaultValue": 3
    },
    {
      "name": "clean.maxPoolSize",
      "type": "java.lang.Integer",
      "description": "最大线程数",
      "sourceType": "etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties",
      "defaultValue": 10
    },
    {
      "name": "clean.queueCapacity",
      "type": "java.lang.Integer",
      "description": "线程池所使用的缓冲队列",
      "sourceType": "etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties",
      "defaultValue": 10000
    },
    {
      "name": "clean.rejectedExecutionHandler",
      "type": "java.lang.String",
      "description": "线程池对拒绝任务的处理策略",
      "sourceType": "etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties",
      "defaultValue": "java.util.concurrent.ThreadPoolExecutor$CallerRunsPolicy"
    }
  ],
  "hints": []
}

3. META-INF下新建 spring-configuration-metadata-whitelist.properties

configuration-properties.classes=etl.dmt.quick.unpackprocessorkafka.config.UnpackProperties

4. 启用

package etl.dmt.quick.unpackprocessorkafka;

import etl.dmt.quick.dto.InputDataDTO;
import etl.dmt.quick.dto.OutputDataDTO;

import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Processor;
import org.springframework.messaging.support.MessageBuilder;


/**
 * @author tianshl
 * @version 2019/8/8 2:46 PM
 */
@EnableBinding(Processor.class)
@EnableConfigurationProperties({UnpackProperties.class})
public class UnpackProcessor {
  ...
}