Spring Cloud Stream Source - 不拉取答案

【问题标题】：Spring Cloud Stream Source - Not PullingSpring Cloud Stream Source - 不拉取
【发布时间】：2020-10-15 08:56:11
【问题描述】：

我正在尝试为 Spring Cloud Dataflow 中的概念验证开发自定义源。

我设法正确部署它，但似乎没有拉出bean。

这是父 pom.xml

的一部分

...
<properties>
  <spring-cloud.version>Hoxton.SR8</spring-cloud.version>
  ...
</properties>

<dependencyManagement>
    <dependencies>
      <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-dependencies</artifactId>
        <version>${spring-cloud.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
    </dependencies>
</dependencyManagement>
...

这是项目 pom.xml

<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-actuator</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-cloud-connectors</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
        <version>2.3.10.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-configuration-processor</artifactId>
        <optional>true</optional>
    </dependency>
</dependencies>

<build>
    <plugins>
        <plugin>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <version>2.3.4.RELEASE</version>
            <configuration>
                <excludes>
                    <exclude>
                        <groupId>org.springframework.boot</groupId>
                        <artifactId>spring-boot-configuration-processor</artifactId>
                    </exclude>
                </excludes>
            </configuration>
        </plugin>
        <plugin>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-app-starter-metadata-maven-plugin</artifactId>
            <version>2.0.2.RELEASE</version>
            <executions>
                <execution>
                    <id>aggregate-metadata</id>
                    <phase>compile</phase>
                    <goals>
                        <goal>aggregate-metadata</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

SourceApplication.java

@SpringBootApplication
@EnableBinding(Source.class)
@EnableConfigurationProperties(ReportingProperties.class)
public class SourceApplication {

    public static void main(String[] args) {
        SpringApplication.run(SourceApplication.class, args);
    }
}

ReportingProperties.java

@Validated
@ConfigurationProperties("reporting-properties")
public class ReportingProperties {

    /**
     * The starting date of the reporting.
     */
    private LocalDateTime fromDate = LocalDateTime.now().minusDays(1000);

    /**
     * The end date of the reporting.
     */
    private LocalDateTime toDate = LocalDateTime.now();

    public LocalDateTime getFromDate() {
        return fromDate;
    }

    public ReportingProperties setFromDate(LocalDateTime fromDate) {
        this.fromDate = fromDate;
        return this;
    }

    public LocalDateTime getToDate() {
        return toDate;
    }

    public ReportingProperties setToDate(LocalDateTime toDate) {
        this.toDate = toDate;
        return this;
    }
}

最后是服务：

@Configuration
@EnableBinding(Source.class)
public class PullUsersService {

    @Bean
    @Publisher(channel = Source.OUTPUT)
    @SendTo(Source.OUTPUT)
    public Supplier<String> pullUsers() {
        return () -> "Test";
    }

}

我想知道如何触发拉动机制，以便在部署时我可以在日志中看到“测试” （我相信 SCDF 上的所有设置都正确，如果我执行“time | log”，我可以在日志中看到一些结果，但如果我执行“myservice | log”，则不会出现任何内容。

我做错了什么？（也许我的代码中有一些冗余）

【问题讨论】：

标签： java spring-kafka spring-cloud-stream spring-cloud-dataflow

【解决方案1】：

上面的答案是正确的，你也可以这样做：

注意：删除所有@EnableBinding(Source.class) 注释

@Component
public class PullUsersService {

    @PollableBean
    public Supplier<String> pullUsers() {
        return () -> "Test";
    }
}

如果你想根据你的 binder 配置排队到 rabbitMq 或 kafka，请添加以下配置

spring.cloud.function.definition=pullUsers
spring.cloud.stream.bindings.pullUsers-out-0.destination=users
spring.cloud.stream.bindings.pullUsers-out-0.group=users-service
spring.cloud.stream.bindings.pullUsers-out-0.producer.requiredGroups=users-service

spring.cloud.stream.bindings.pullUsers-out-0.producer.requiredGroups 配置强制生产者创建队列

【讨论】：

【解决方案2】：

让你使用它的原因很有趣：

@Publisher(channel = Source.OUTPUT)
@SendTo(Source.OUTPUT)

如果您查看您提到的 time 源代码，您会看到如下内容：

@PollableSource
public String publishTime() {
    return new SimpleDateFormat(this.triggerProperties.getDateFormat()).format(new Date());
}

考虑改用@PollableSource，不要使用Supplier。关键是您当前的所有注释都与轮询无关。

@Publisher 仅在我们调用该方法时有效。 @SendTo 在此处被完全忽略，因为 @Publisher 的作用完全相同，并且它“发送到”。

【讨论】：