时间戳

首页 > 解决方案 > 将 Spring 云流应用程序 - Chelsea 迁移到 Horsham

spring-cloud-stream - 将 Spring 云流应用程序 - Chelsea 迁移到 Horsham

问题描述

我正在尝试将 Spring Cloud Stream 应用程序 Chelsea.SR1 迁移到 Horsham。该项目没有 binder，但使用 spring 消息传递和集成来集成自定义源和接收器。

看起来AggregateApplicationBuilder不再可用了。如何在没有 的情况下绑定源、变压器和接收器AggregateApplicationBuilder？

或者有没有更好的方法在 Horsham 中实现这一点？我唯一的限制是，我需要使用可轮询的源/消费者。

提前致谢！

pom.xml

  <modelVersion>4.0.0</modelVersion>
  <artifactId>example-info-processor</artifactId>
  <groupId>com.example.streams</groupId>
  <version>1.2-SNAPSHOT</version>
  <packaging>jar</packaging>
  <parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.9.RELEASE</version>
    <relativePath />
  </parent>
  <properties>
    <spring-cloud.version>Chelsea.SR1</spring-cloud.version>
  </properties>
  <dependencies>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-stream</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-messaging</artifactId>
      <version>4.3.17.RELEASE</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-core</artifactId>
      <version>4.3.15.RELEASE</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-web</artifactId>
      <version>4.3.20.RELEASE</version>
    </dependency>
    <dependency>
      <groupId>org.springframework</groupId>
      <artifactId>spring-webmvc</artifactId>
      <version>4.3.24.RELEASE</version>
    </dependency>
  </dependencies>
  <dependencyManagement>
    <dependencies>
      <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-dependencies</artifactId>
        <version>${spring-cloud.version}</version>
        <type>pom</type>
        <scope>import</scope>
      </dependency>
    </dependencies>
  </dependencyManagement>
</project>

应用

@SpringBootApplication
@ComponentScan(basePackages = {
        "com.example.streams",
})
public class ExampleInfoProcessorApplication extends SpringBootServletInitializer {
    public static void main(String[] args) {
        new AggregateApplicationBuilder(ExampleInfoProcessor.class, args)
                .from(ExampleInfoConsumer.class).via(ExampleInfoTransformer.class)
                .to(ExampleInfoSink.class).run(args);
    }
}

资源

@EnableBinding(Source.class)
public class ExampleInfoConsumer {
    @Autowired
    private MyConsumer customConsumer;

    @InboundChannelAdapter(value = Source.OUTPUT, poller = @Poller(fixedDelay = "30"))
    public List<ExampleRecord> consume() {
      //Read the available chunk from the source location
            .....
        return exampleRecordsList;
    }
}

变压器

@EnableBinding(Processor.class)
public class ExampleInfoTransformer {

    @Transformer(inputChannel = Processor.INPUT, outputChannel = Processor.OUTPUT)
    public Map<String, List<ExampleInfo>> transform(List<ExampleRecord> exampleRecords) {
        //transform the consumer records, store the list of ExampleInfo objects in a map using ExampleInfo key
        return map;
    }
}

下沉

@EnableBinding(Sink.class)
public class ExampleInfoSink {

    @ServiceActivator(inputChannel = Sink.INPUT)
    public void sink(Map<String, List<ExampleInfo>>  exampleInfoMap) {
        //Read from map and write to destination.
    }
}

标签： spring-cloud-stream

解决方案

如果你只使用 spring-integration 和 spring-messaging 为什么你有 spring-cloud-stream 作为依赖而不是 spring-integration 这是一个顶级框架？如何在没有的情况下绑定源、变压器和接收器。. . - 源有输出通道，变压器有输入通道。只要输出和输入是相同的通道，您的组件就会连接。但这与 spring-cloud-stream 无关。那是纯粹的弹簧集成。

Chelsea 是一个非常旧的版本（1.x）。我们已AggregateApplicationBuilder在 2.x 中弃用并在 3.x 中将其删除，以支持功能组合。我们也不再提倡 spring-cloud-stream 的基于注释的配置模型，而是支持基于函数的编程模型。

请阅读我们最近发布的以下帖子：

推荐阅读