知識庫 / Spring / Spring Cloud RSS 訂閱

@EnableBinding(Source.class)
@SpringBootApplication
public class SpringDataFlowTimeSourceApplication {
    
    public static void main(String[] args) {
        SpringApplication.run(
          SpringDataFlowTimeSourceApplication.class, args);
    }
}

現在我們需要定義需要處理的數據源。該數據源可以是任何潛在的無限工作負載，例如物聯網傳感器數據、7x24小時事件處理或在線交易數據導入。

在我們的示例應用程序中，我們使用一個Poller，每 10 秒產生一個事件（為了簡化，新時間戳）。

@InboundChannelAdapter 註解將消息發送到源的輸出通道，並將返回值作為消息的負載：

@Bean
@InboundChannelAdapter(
  value = Source.OUTPUT, 
  poller = @Poller(fixedDelay = "10000", maxMessagesPerPoll = "1")
)
public MessageSource<Long> timeMessageSource() {
    return () -> MessageBuilder.withPayload(new Date().getTime()).build();
}

我們的數據源已準備就緒。

7. 處理器應用程序

接下來，我們將創建一個應用程序並添加 Stream Rabbit 依賴項。

然後，我們將向 Spring Boot 主類添加 @EnableBinding(Processor.class) 註解：

@EnableBinding(Processor.class)
@SpringBootApplication
public class SpringDataFlowTimeProcessorApplication {

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

接下來，我們需要定義一個方法來處理來自源應用程序的數據。

為了定義一個轉換器，我們需要使用 @Transformer 註解對該方法進行標註：

@Transformer(inputChannel = Processor.INPUT, 
  outputChannel = Processor.OUTPUT)
public Object transform(Long timestamp) {

    DateFormat dateFormat = new SimpleDateFormat("yyyy/MM/dd hh:mm:yy");
    String date = dateFormat.format(timestamp);
    return date;
}

它將來自“input”通道的時間戳轉換為格式化的日期，並將該日期發送到“output”通道。

8. Sink 應用程序

最後創建的應用程序是 Sink 應用程序。

再次，前往 Spring Initializr 並選擇一個 Group 和一個 Artifact 名稱。 下載項目後，添加一個 Stream Rabbit 依賴項。

然後，將 @EnableBinding(Sink.class) 註解添加到 Spring Boot 主類中：

@EnableBinding(Sink.class)
@SpringBootApplication
public class SpringDataFlowLoggingSinkApplication {

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

現在我們需要一種方法來攔截來自處理器的消息。

要做到這一點，我們需要將 @StreamListener(Sink.INPUT) 註解添加到我們的方法中：

@StreamListener(Sink.INPUT)
public void loggerSink(String date) {
    logger.info("Received: " + date);
}

該方法僅將格式化後的時間戳打印到日誌文件中。

9. 註冊流應用

Spring Cloud Data Flow Shell 允許我們使用 app register 命令與應用註冊表註冊流應用。

我們必須提供一個唯一的名稱、應用程序類型和可以解析為應用程序 artifact 的 URI。對於類型，請指定 "source“、"processor“ 或 "sink“。

當提供使用 maven 方案的 URI 時，格式應符合以下要求：

maven://<groupId>:<artifactId>[:<extension>[:<classifier>]]:<version>

要註冊之前創建的 Source, Processor, 和 Sink 應用程序，請轉到 Spring Cloud Data Flow Shell，並在提示符下執行以下命令：

app register --name time-source --type source --uri file://local machine path to/time-source-0.0.1-SNAPSHOT.jar

app register --name time-processor --type processor --uri file://local machine path to/time-processor-0.0.1-SNAPSHOT.jar

app register --name log-sink --type sink --uri file://local machine path to/log-sink-0.0.1-SNAPSHOT.jar

10. 創建並部署流

要創建新的流定義，請轉到 Spring Cloud Data Flow Shell，並執行以下 shell 命令:

stream create --name time-to-log --definition 'time-source | time-processor | log-sink'

這定義了一個名為 time-to-log 的流，基於 DSL 表達式 ‘time-source | time-processor | log-sink’。

要部署該流，請執行以下 shell 命令：

stream deploy --name time-to-log

Data Flow Server 會解析時間源、時間處理器和日誌接收器到 Maven 座標，並利用這些座標啓動流中的時間源、時間處理器和日誌接收器應用。

如果流已正確部署，您將在Data Flow Server 的日誌中看到模塊已啓動並連接在一起：

2024-04-15 15:15:27.153  INFO 23568 — [pool-3-thread-1] o.s.c.s.s.d.DefaultReleaseManager        : Getting status for org.springframework.cloud.skipper.domain.Release@4de8f9c3 using deploymentIds time-to-log2.time-processor-v1,time-to-log2.time-source1-v1,time-to-log2.log-sink-v1
2024-04-15 15:15:32.156  INFO 23568 — [pool-3-thread-1] o.s.c.s.s.d.DefaultReleaseManager        : Getting status for org.springframework.cloud.skipper.domain.Release@4d72121e using deploymentIds time-to-log2.time-processor-v1,time-to-log2.time-source1-v1,time-to-log2.log-sink-v1
2024-04-15 15:15:37.157  INFO 23568 — [pool-3-thread-1] o.s.c.s.s.d.DefaultReleaseManager        : Getting status for org.springframework.cloud.skipper.domain.Release@9a35173 using deploymentIds time-to-log2.time-processor-v1,time-to-log2.time-source1-v1,time-to-log2.log-sink-v1
2024-04-15 15:15:42.160  INFO 23568 — [pool-3-thread-1] o.s.c.s.s.d.DefaultReleaseManager        : Getting status for org.springframework.cloud.skipper.domain.Release@1ec5d911 using deploymentIds time-to-log2.time-processor-v1,time-to-log2.time-source1-v1,time-to-log2.log-sink-v1

11. 結果審查

在本示例中，源端每秒將當前時間戳發送為消息，處理器對其進行格式化，日誌接收器則使用日誌框架輸出格式化的時間戳。

日誌文件位於 的日誌輸出中，如上所示。

tail -f PATH_TO_LOG/spring-cloud-dataflow-1276836171391672089/time-to-log-1472034549734/time-to-log.logging-sink/stdout_0.log
2016-08-24 12:40:42.029  INFO 9488 --- [r.time-to-log-1] s.c.SpringDataFlowLoggingSinkApplication : Received: 2016/08/24 11:40:01
2016-08-24 12:40:52.035  INFO 9488 --- [r.time-to-log-1] s.c.SpringDataFlowLoggingSinkApplication : Received: 2016/08/24 11:40:11
2016-08-24 12:41:02.030  INFO 9488 --- [r.time-to-log-1] s.c.SpringDataFlowLoggingSinkApplication : Received: 2016/08/24 11:40:21