知識庫 / Spring RSS 訂閱

1. 簡介

本文將主要通過小而實用的示例，介紹 Spring Integration 的核心概念。

Spring Integration 提供大量的強大組件，可以極大地增強企業架構中系統和流程的互聯互通。

它體現了一些最優秀和最流行的設計模式，幫助開發者避免自行實現。

我們將探討該庫所滿足的具體需求，以及為什麼在某些替代方案中，選擇它是有益的。我們還將探討用於進一步簡化 Spring Integration 應用程序開發的可用工具。

2. 設置

<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-core</artifactId>
    <version>6.0.0</version>
</dependency>
<dependency>
    <groupId>org.springframework.integration</groupId>
    <artifactId>spring-integration-file</artifactId>
    <version>6.0.0</version>
</dependency>

您可以在 Maven Central 找到最新版本的 Spring Integration Core 和 Spring Integration File Support。

3. 消息傳遞模式

本庫中的一個基石模式是消息傳遞。該模式圍繞消息展開——數據離散負載，通過預定義的通道從一個或多個系統或進程中向系統或進程移動。

該模式的歷史起源是作為一種最靈活的方式來集成多個異構系統，以實現以下目標：

• 幾乎完全解耦參與集成中的系統
• 允許集成中的參與系統對彼此底層協議、格式或其他實現細節完全無知
• 鼓勵參與集成中組件的開發和重用

4. 消息集成示例

讓我們考慮一個基本的示例，該示例將一個 MPEG 視頻文件從一個指定文件夾複製到另一個配置的文件夾：

@Configuration
@EnableIntegration
public class BasicIntegrationConfig{
    public String INPUT_DIR = "the_source_dir";
    public String OUTPUT_DIR = "the_dest_dir";
    public String FILE_PATTERN = "*.mpeg";

    @Bean
    public MessageChannel fileChannel() {
        return new DirectChannel();
    }

    @Bean
    @InboundChannelAdapter(value = "fileChannel", poller = @Poller(fixedDelay = "1000"))
    public MessageSource<File> fileReadingMessageSource() {
        FileReadingMessageSource sourceReader= new FileReadingMessageSource();
        sourceReader.setDirectory(new File(INPUT_DIR));
        sourceReader.setFilter(new SimplePatternFileListFilter(FILE_PATTERN));
        return sourceReader;
    }

    @Bean
    @ServiceActivator(inputChannel= "fileChannel")
    public MessageHandler fileWritingMessageHandler() {
        FileWritingMessageHandler handler = new FileWritingMessageHandler(new File(OUTPUT_DIR));
        handler.setFileExistsMode(FileExistsMode.REPLACE);
        handler.setExpectReply(false);
        return handler;
    }
}

上述代碼配置了一個服務激活器、集成通道和一個入站通道適配器。

我們稍後將詳細研究這些組件類型。@EnableIntegration註解將此類標記為Spring Integration配置。

現在，讓我們開始我們的Spring Integration應用程序上下文：

public static void main(String... args) {
    AbstractApplicationContext context 
      = new AnnotationConfigApplicationContext(BasicIntegrationConfig.class);
    context.registerShutdownHook();
    
    Scanner scanner = new Scanner(System.in);
    System.out.print("Please enter q and press <enter> to exit the program: ");
    
    while (true) {
       String input = scanner.nextLine();
       if("q".equals(input.trim())) {
          break;
      }
    }
    System.exit(0);
}

<p>主方法以上啓動集成上下文；它還接受命令行中的“<em >q</em>”字符輸入以退出程序。讓我們更詳細地考察一下組件。</p>

5. Spring Integration 組件

5.1. 消息

<em style="font-size: 16px;line-height: 1.5">org.springframework.integration.Message</em> 接口定義了 Spring Message：在 Spring Integration 環境中，數據的傳輸單元。

public interface Message<T> {
    T getPayload();
    MessageHeaders getHeaders();
}

它定義了兩個關鍵元素的訪問器：

• 消息頭，本質上是一個鍵值容器，可用於傳輸元數據，如定義在 org.springframework.integration.MessageHeaders 類中的
• 消息負載，即具有傳輸價值的實際數據——在我們的用例中，視頻文件是負載

5.2. 通道

在 Spring Integration（以及 EAI 領域）中，通道是集成架構的基本“管道”。它用於將消息從一個系統傳遞到另一個系統。

你可以將其視為一個字面上的管道，通過它可以集成系統或流程將消息推送到（或從）其他系統。

Spring Integration 中的通道有多種類型，具體取決於你的需求。它們 largely 可配置且可直接使用，無需編寫任何自定義代碼，但如果你的需求是定製的，則有一個強大的框架可用。

點對點 (P2P) 通道 用於在系統或組件之間建立 1-對-1 的通信線路。一個組件將消息發佈到通道，以便另一個組件可以將其拾取。每個通道的末端只能有一個組件。

正如我們所見，配置通道就像返回一個 DirectChannel 的實例一樣簡單。

@Bean
public MessageChannel fileChannel1() {
    return new DirectChannel();
}

@Bean
public MessageChannel fileChannel2() {
    return new DirectChannel();
}

@Bean
public MessageChannel fileChannel3() {
    return new DirectChannel();
}

在這裏，我們定義了三個獨立的通道，它們都通過各自的 getter 方法的名稱進行標識。

發佈-訂閲 (Pub-Sub) 通道用於在系統或組件之間建立一對多通信線路。這將允許我們發佈到我們之前創建的這三個直接通道。

因此，根據我們的示例，我們可以用 pub-sub 通道替換 P2P 通道：

@Bean
public MessageChannel pubSubFileChannel() {
    return new PublishSubscribeChannel();
}

@Bean
@InboundChannelAdapter(value = "pubSubFileChannel", poller = @Poller(fixedDelay = "1000"))
public MessageSource<File> fileReadingMessageSource() {
    FileReadingMessageSource sourceReader = new FileReadingMessageSource();
    sourceReader.setDirectory(new File(INPUT_DIR));
    sourceReader.setFilter(new SimplePatternFileListFilter(FILE_PATTERN));
    return sourceReader;
}

我們現在已將 inbound 渠道適配器轉換為發佈到 Pub-Sub 渠道。這將允許我們將從源文件夾中讀取的文件發送到多個目的地。

5.3. 橋樑 (Bridge)

橋樑 (Bridge) 在 Spring Integration 中用於連接兩個消息通道或適配器，如果它們無法直接連接，則使用橋樑。

在我們的案例中，我們可以使用橋樑來連接我們的 Pub-Sub 通道與三個不同的 P2P 通道（因為 P2P 通道和 Pub-Sub 通道無法直接連接）。

@Bean
@BridgeFrom(value = "pubSubFileChannel")
public MessageChannel fileChannel1() {
    return new DirectChannel();
}

@Bean
@BridgeFrom(value = "pubSubFileChannel")
public MessageChannel fileChannel2() {
    return new DirectChannel();
}

@Bean
@BridgeFrom(value = "pubSubFileChannel")
public MessageChannel fileChannel3() {
    return new DirectChannel();
}

上述 Bean 配置現在將 pubSubFileChannel 橋接到三個 P2P 頻道。 @BridgeFrom 註解定義了橋接，可以應用於需要訂閲 Pub-Sub 頻道並且數量不限的任何頻道。

我們可以將上述代碼解讀為“創建一個從 pubSubFileChannel 到 fileChannel1, fileChannel2, 和 fileChannel3 的橋接，以便 Pub-Sub 頻道中的消息可以同時傳遞到這三個頻道。”

5.4. 服務激活器

服務激活器是指定義了 <em @ServiceActivator 標註的任何 POJO。這允許我們在接收到 inbound 渠道的消息時執行 POJO 中的任何方法，並允許我們向 outward 渠道寫入消息。

在我們的示例中，服務激活器接收來自配置的 <em input channel 的文件，並將其寫入配置的文件夾。

5.5. 適配器

適配器是一種基於企業集成模式的組件，允許您連接到系統或數據源。它幾乎字面意義上就是我們從插座或電子設備中拔插時所説的適配器。

它允許您連接到其他“黑盒”系統，例如數據庫、FTP 服務器和消息系統，如 JMS、AMQP 和社交網絡（如 Twitter）。連接到這些系統的普遍需求意味着適配器具有很高的可移植性和可重用性（事實上，有一個小型 適配器目錄，免費供任何人使用）。

適配器可以分為兩大類：入站和出站。

讓我們在樣本場景中分析這些類別：

入站適配器用於從外部系統（在本例中為文件系統目錄）獲取消息。

我們的入站適配器配置包括：

• 一個 ，標記 bean 配置為適配器——我們配置適配器將消息饋送到的通道（在本例中為 MPEG 文件）以及一個 poller，一個組件，它幫助適配器以指定的時間間隔輪詢配置的文件夾
• 一個標準的 Spring java 配置類，返回一個 FileReadingMessageSource，Spring Integration 類實現，它處理文件系統輪詢