動態

詳情 返回 返回

Spring Integration 輕鬆實現服務間消息傳遞，真香！ - 動態 詳情

在當今分佈式系統的背景下，如何優雅地實現系統之間的消息傳遞是每個開發者都關心的話題。而Spring Integration，作為Spring家族的一員，正是為了解決這個難題而生。

在這篇文章中，我們將踏上穿越消息之路，深入探討Spring Integration的魅力。

關注公眾號：碼猿技術專欄，回覆關鍵詞：1111 獲取阿里內部Java性能調優手冊！

Spring Integration基礎概念

1. 起源：

  • Spring Integration是Spring框架的一個擴展，旨在簡化企業集成模式的開發。它提供了一種基於消息的編程模型，使得在分佈式系統中進行系統集成變得更加容易。

2. 基本概念：

  • 消息： Spring Integration使用消息來在系統中傳遞信息。消息是信息的載體，它可以包含業務數據、頭部信息、消息標籤等。消息在系統中沿着通道（Channel）傳遞。
  • 通道（Channel）： 通道是消息在系統中傳遞的管道。Spring Integration提供了不同類型的通道，如直接通道（Direct Channel）、發佈-訂閲通道（Publish-Subscribe Channel）、隊列通道（Queue Channel）等。
  • 端點（Endpoint）： 端點是消息的生產者或者消費者。消息從一個端點流向另一個端點，形成一個消息的處理流程。
  • 適配器（Adapter）： 適配器用於將外部系統或者服務與Spring Integration整合。它可以將外部系統的消息轉換為Spring Integration的消息，也可以將Spring Integration的消息傳遞給外部系統。
  • 過濾器（Filter）： 過濾器用於過濾消息，只有滿足特定條件的消息才能通過。它可以用於消息的路由、轉換等。
  • 轉換器（Transformer）： 轉換器用於將消息從一種形式轉換為另一種形式，以滿足系統的需求。它可以用於數據格式的轉換、消息體的修改等。
推薦下博主的小冊子，企業級實戰總結40講

Spring Integration與傳統消息中間件的區別與聯繫：

1. 區別：

  • Spring Integration是框架： Spring Integration是一個基於Spring的框架，它提供了一整套用於構建企業集成模式的工具和組件。
  • 傳統消息中間件是產品： 傳統消息中間件通常是獨立的產品，如RabbitMQ、Apache Kafka、ActiveMQ等，它們專注於提供消息傳遞服務。

2. 聯繫：

  • 整合性： Spring Integration可以與傳統消息中間件集成使用，通過適配器與外部消息中間件進行通信。這樣，Spring Integration可以作為一箇中間層，幫助企業集成系統與不同的消息中間件進行對接。
  • 解耦與異步通信： 類似傳統消息中間件，Spring Integration也支持解耦和異步通信的模式，通過消息的發佈與訂閲，實現系統組件之間的解耦和鬆耦合。
  • 消息傳遞： Spring Integration和傳統消息中間件一樣，都是基於消息傳遞的模型。消息作為信息的載體，在系統中傳遞，實現不同組件之間的通信。

總體而言，Spring Integration提供了一種更加輕量級和靈活的方式來實現企業集成，而傳統消息中間件更專注於提供可靠的消息傳遞服務。在實際應用中，可以根據具體的需求選擇合適的技術和工具。

關注公眾號：碼猿技術專欄，回覆關鍵詞：1111 獲取阿里內部Java性能調優手冊！

消息通道與消息端點

消息通道與消息端點：

定義和配置消息通道：

  1. 定義消息通道：
    • 在Spring Integration中，消息通道是消息在系統中傳遞的管道。可以使用XML配置或Java代碼來定義消息通道。
    • XML配置示例：
<int:channel id="myChannel"/>
    • Java配置示例：
@Bean
public MessageChannel myChannel() {
    return MessageChannels.direct().get();
}

1.配置消息通道的類型：

    • Spring Integration提供了不同類型的消息通道，如直接通道（Direct Channel）、發佈-訂閲通道（Publish-Subscribe Channel）、隊列通道（Queue Channel）等。可以根據需求選擇合適的通道類型。
    • XML配置示例：
<!-- 配置直接通道 -->
<int:channel id="directChannel"/>
<!-- 配置發佈-訂閲通道 -->
<int:publish-subscribe-channel id="publishSubscribeChannel"/>
<!-- 配置隊列通道 -->
<int:queue-channel id="queueChannel"/>
    • Java配置示例：
@Bean
public MessageChannel directChannel() {
    return MessageChannels.direct().get();
}
@Bean
public MessageChannel publishSubscribeChannel() {
    return MessageChannels.publishSubscribe().get();
}
@Bean
public MessageChannel queueChannel() {
    return MessageChannels.queue().get();

2.消息通道的屬性配置：

    • 可以通過配置消息通道的一些屬性，如容量、過期時間等，以滿足具體的需求。
    • XML配置示例：
<int:channel id="myChannel" capacity="10" />
    • Java配置示例：
@Bean
public MessageChannel myChannel() {
    return MessageChannels.direct().capacity(10).get();
}

消息端點的作用和類型：

  1. 作用：
    • 消息端點是消息的生產者或者消費者，它定義了消息的處理邏輯。消息從一個端點流向另一個端點，形成一個消息的處理流程。

2.消息端點的類型：

    • 消息生產者端點：
      • 消息源（Message Source）： 用於產生消息的端點，如文件輸入、JDBC查詢等。
      • 通道適配器（Channel Adapter）： 用於將外部系統的消息轉換為Spring Integration的消息格式。
    • 消息消費者端點：
      • 服務激活器（Service Activator）： 用於將消息傳遞給特定的服務進行處理。
      • 消息處理器（Message Handler）： 用於處理消息，可以是一個Java方法、表達式、腳本等。
    • 消息路由器端點：
      • 分發器（Dispatcher）： 用於將消息分發給不同的子通道，根據條件進行消息路由。
    • 其他類型：
      • 過濾器（Filter）： 用於過濾消息，只有滿足特定條件的消息才能通過。
      • 轉換器（Transformer）： 用於將消息從一種形式轉換為另一種形式。

3.配置消息端點：

    • 消息端點可以通過XML配置或Java代碼進行定義。
    • XML配置示例：
<int:service-activator input-channel="myChannel" ref="myService" method="processMessage"/>
    • Java配置示例：
@ServiceActivator(inputChannel = "myChannel")
public void processMessage(Message<String> message) {
    // 處理消息的邏輯
}

通過合理定義和配置消息通道以及消息端點，可以構建出靈活、可擴展的消息傳遞系統，實現消息在系統中的流動和處理。

消息處理器與適配器

消息處理器與適配器在Spring Integration中的使用：

1. 消息處理器的使用方法：

消息處理器是Spring Integration中用於處理消息的組件，它可以是一個Java方法、表達式、腳本等。以下是消息處理器的使用方法：

  • Java方法處理器：
@ServiceActivator(inputChannel = "inputChannel")
public void handleMessage(String message) {
    // 處理消息的邏輯
    System.out.println("Received Message: " + message);
}
  • 上述代碼中，handleMessage方法是一個消息處理器，通過@ServiceActivator註解將其與名為inputChannel的輸入通道關聯起來。當消息被髮送到該通道時，該方法會被調用來處理消息。
  • 表達式處理器：
<int:service-activator input-channel="inputChannel" expression="@myService.process(#payload)">
    <int:poller fixed-rate="1000"/>
</int:service-activator>
  • 上述配置中，expression屬性定義了一個表達式，指定了消息處理的邏輯。這個表達式將調用名為process的方法，#payload表示消息的載荷。

2. 適配器與外部系統集成：

適配器用於將外部系統的消息與Spring Integration進行集成，使得外部系統的消息能夠在Spring Integration中流通。以下是適配器的使用方法：

  • 文件適配器：
<int-file:inbound-channel-adapter id="filesIn"
                                 channel="inputChannel"
                                 directory="file:${java.io.tmpdir}/input">
    <int:poller fixed-rate="5000"/>
</int-file:inbound-channel-adapter>
  • 上述配置使用文件適配器（<int-file:inbound-channel-adapter>）來監聽指定目錄中的文件，並將文件內容發送到名為inputChannel的通道。
  • JDBC適配器：
<int-jdbc:inbound-channel-adapter id="jdbcInboundAdapter"
                                  query="SELECT * FROM my_table"
                                  channel="inputChannel">
    <int:poller fixed-rate="10000"/>
</int-jdbc:inbound-channel-adapter>
  • 上述配置中，JDBC適配器（<int-jdbc:inbound-channel-adapter>）從數據庫執行查詢，並將結果發送到inputChannel通道。
  • HTTP適配器：
<int-http:inbound-channel-adapter id="httpInboundAdapter"
                                  channel="inputChannel"
                                  path="/receiveMessage"
                                  request-mapper="requestMapping">
    <int:poller fixed-rate="10000"/>
</int-http:inbound-channel-adapter>
  • 上述配置使用HTTP適配器（<int-http:inbound-channel-adapter>）監聽指定路徑的HTTP請求，並將請求的消息發送到inputChannel通道。

以上示例展示瞭如何使用不同類型的適配器來與外部系統進行集成。適配器將外部系統的消息轉換為Spring Integration的消息，並通過通道在整個系統中傳遞。適配器的配置取決於具體的集成需求和外部系統的特性。

消息轉換與路由在Spring Integration中的應用

1. 消息的格式轉換與處理：

消息轉換是Spring Integration中常見的操作，用於將消息從一種格式或結構轉換為另一種格式或結構，以滿足系統的需求。以下是消息轉換的實際應用場景和示例：

  • JSON到對象的轉換：
@Transformer(inputChannel = "jsonInputChannel", outputChannel = "objectOutputChannel")
public MyObject convertJsonToObject(String jsonString) {
    // 使用Jackson庫將JSON字符串轉換為Java對象
    return objectMapper.readValue(jsonString, MyObject.class);
}
  • 上述代碼中，@Transformer註解表示這是一個消息轉換器，將jsonInputChannel通道的JSON消息轉換為Java對象，並將結果發送到objectOutputChannel通道。
  • 對象到JSON的轉換：
@Transformer(inputChannel = "objectInputChannel", outputChannel = "jsonOutputChannel")
public String convertObjectToJson(MyObject myObject) {
    // 使用Jackson庫將Java對象轉換為JSON字符串
    return objectMapper.writeValueAsString(myObject);
}
  • 在這個例子中，消息轉換器將objectInputChannel通道的Java對象轉換為JSON字符串，並將結果發送到jsonOutputChannel通道。

2. 路由器的作用和實際應用場景：

路由器用於根據消息的內容或特徵將消息路由到不同的通道，實現消息在系統中的分發。以下是路由器的實際應用場景和示例：

  • 內容路由器：
<int:router input-channel="inputChannel" expression="payload.type">
    <int:mapping value="A" channel="channelA"/>
    <int:mapping value="B" channel="channelB"/>
    <int:mapping value="C" channel="channelC"/>
</int:router>
  • 上述配置中，內容路由器（<int:router>）根據消息的type屬性的值將消息路由到不同的通道。如果消息的type是"A"，則路由到channelA；如果是"B"，則路由到channelB，以此類推。
  • 篩選器路由器：
<int:router input-channel="inputChannel">
    <int:mapping value="payload.type == 'A'" channel="channelA"/>
    <int:mapping value="payload.type == 'B'" channel="channelB"/>
    <int:mapping value="payload.type == 'C'" channel="channelC"/>
</int:router>
  • 在這個例子中，路由器根據篩選條件將消息路由到不同的通道。只有滿足條件的消息才會被路由到相應的通道。

路由器的靈活性使得可以根據消息的內容、屬性或條件進行動態的路由，從而實現系統中不同組件的消息處理邏輯的分離。路由器的配置可以根據具體的需求進行調整，以適應不同的應用場景。

集成模式與設計模式

Spring Integration中常見的集成模式：

Spring Integration提供了許多常見的集成模式，這些模式幫助開發人員構建可靠、可擴展的消息驅動系統。以下是一些常見的集成模式：

  1. 消息通道（Message Channel）：
    • 定義了消息在系統中傳遞的路徑，是消息傳遞的媒介。

2.消息端點（Message Endpoint）：

    • 定義了消息的生產者或者消費者，可以是服務激活器、消息處理器等。

3.消息適配器（Message Adapter）：

    • 用於將外部系統的消息轉換為Spring Integration的消息格式，實現系統與外部系統的集成。

4.消息網關（Message Gateway）：

    • 提供了對系統的入口，允許外部系統通過網關發送消息到系統中，或者從系統中獲取消息。

5.消息轉換器（Message Transformer）：

    • 用於對消息的格式進行轉換，將消息從一種表示形式轉換為另一種，以滿足系統的需求。

6.消息過濾器（Message Filter）：

    • 用於過濾消息，只有滿足特定條件的消息才能通過，實現對消息的篩選。

7.消息路由器（Message Router）：

    • 根據消息的內容、屬性或條件將消息路由到不同的通道，實現消息的分發。

8.聚合器（Aggregator）：

    • 將多個相關的消息合併為一個消息，通常用於處理分散的消息片段。

9.分裂器（Splitter）：

    • 將一個消息拆分為多個消息，通常用於處理大塊的消息內容。

10.定時器（Timer）：

    • 定期發送消息，用於實現定時任務或者輪詢外部系統。

如何根據設計模式構建消息驅動的系統：

構建消息驅動的系統時，可以借鑑一些設計模式來提高系統的可維護性、可擴展性和可測試性。以下是一些常用的設計模式，特別是在消息驅動系統中的應用：

  1. 發佈-訂閲模式（Publish-Subscribe Pattern）：
    • 在消息驅動系統中，通過使用發佈-訂閲模式可以實現消息的廣播，允許多個組件訂閲並接收相同的消息。

2.觀察者模式（Observer Pattern）：

    • 觀察者模式可以用於實現消息的訂閲和通知機制，在消息產生時通知所有的觀察者。

3.策略模式（Strategy Pattern）：

    • 策略模式可用於實現靈活的消息處理策略，根據不同的需求選擇不同的消息處理算法。

4.裝飾者模式（Decorator Pattern）：

    • 裝飾者模式可用於動態地添加消息處理邏輯，如消息轉換器、消息過濾器等。

5.責任鏈模式（Chain of Responsibility Pattern）：

    • 責任鏈模式可用於實現消息處理管道，每個處理器負責處理特定類型的消息，形成一個處理鏈。

6.命令模式（Command Pattern）：

    • 命令模式可以將消息封裝為命令對象，以支持撤銷、重做等操作。

7.工廠模式（Factory Pattern）：

    • 工廠模式可用於創建消息適配器、消息處理器等組件，提供一種靈活的對象創建方式。

通過結合這些設計模式，可以更好地組織和管理消息驅動系統的代碼，使系統更易於擴展和維護。選擇適當的設計模式取決於系統的特定需求和架構。

Spring Integration中流程和通道攔截的實現方法

在Spring Integration中，可以通過攔截器（Interceptor）來對消息通道和流程進行攔截和處理。攔截器允許在消息在通道中傳遞和處理的過程中執行自定義邏輯。

1. 通道攔截：

在通道級別，可以使用通道攔截器來對消息通道的發送和接收進行攔截。

<int:channel id="myChannel">
    <int:interceptors>
        <int:wire-tap channel="logChannel"/>
    </int:interceptors>
</int:channel>

上述配置中，<int:wire-tap>是一個通道攔截器，將通道上的所有消息發送到logChannel通道，以便記錄日誌或進行其他操作。

2. 流程攔截：

在流程級別，可以使用<int:advice>和<int:expression-advice>等元素來添加攔截器。

<int:service-activator input-channel="inputChannel" output-channel="outputChannel">
    <int:advice-chain>
        <int:expression-advice expression="payload.toUpperCase()"/>
    </int:advice-chain>
</int:service-activator>

在上述配置中，<int:expression-advice>是一個流程攔截器，它使用SpEL表達式將消息內容轉換為大寫。

攔截器的應用和自定義：

1. 內置攔截器的應用：

Spring Integration提供了一些內置的攔截器，如WireTap、LoggingHandler等，用於實現常見的攔截需求。例如：

<int:channel id="inputChannel">
    <int:interceptors>
        <int:wire-tap channel="logChannel"/>
    </int:interceptors>
</int:channel>

上述配置中，使用了內置的WireTap攔截器，將通道上的所有消息發送到logChannel通道。

2. 自定義攔截器：

可以通過實現ChannelInterceptor接口或擴展ChannelInterceptorAdapter類來創建自定義的通道攔截器。同樣，通過實現Advice接口或擴展AbstractRequestHandlerAdvice類可以創建自定義的流程攔截器。

public class CustomChannelInterceptor implements ChannelInterceptor {
    @Override
    public Message<?> preSend(Message<?> message, MessageChannel channel) {
        // 在消息發送之前執行的邏輯
        return message;
    }
    
    @Override
    public void afterSendCompletion(Message<?> message, MessageChannel channel, boolean sent, Exception ex) {
        // 在消息發送完成後執行的邏輯
    }
    
    // 其他方法省略
}

<int:service-activator input-channel="inputChannel" output-channel="outputChannel">
    <int:advice-chain>
        <bean class="com.example.CustomExpressionAdvice"/>
    </int:advice-chain>
</int:service-activator>

上述配置中，使用了自定義的流程攔截器CustomExpressionAdvice，該類需實現Advice接口。

通過應用內置或自定義的攔截器，可以在消息處理的不同階段執行自定義的邏輯，如日誌記錄、性能監控、消息轉換等。

實戰

傳統訂單處理流程往往涉及多個手動步驟，容易導致延遲和錯誤。為了提高電商平台的運作效率，客户那邊要求我們開發一個自動化訂單處理系統，從訂單創建到支付、庫存檢查和發貨全流程自動化處理，通過消息觸發相關的業務邏輯，減少人為失誤。

1.添加依賴：

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-integration</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.projectlombok</groupId>
            <artifactId>lombok</artifactId>
            <optional>true</optional>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-test</artifactId>
            <scope>test</scope>
        </dependency>

2.啓動類Application：

@SpringBootApplication
@IntegrationComponentScan
public class OrderProcessingApplication {

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

3.配置消息通道

/**
 * 配置消息通道
 */
@Configuration
public class IntegrationConfig {

    /**
     * 定義訂單創建的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel orderCreatedChannel() {
        return new DirectChannel();
    }

    /**
     * 定義支付處理的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel paymentProcessedChannel() {
        return new DirectChannel();
    }

    /**
     * 定義庫存檢查的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel inventoryCheckedChannel() {
        return new DirectChannel();
    }

    /**
     * 定義發貨調度的消息通道
     * @return DirectChannel 實例
     */
    @Bean
    public MessageChannel shipmentScheduledChannel() {
        return new DirectChannel();
    }
}

4.Controller

@RestController
@RequestMapping("/orders")
public class OrderController {

    private final OrderService orderService;

    @Autowired
    public OrderController(OrderService orderService) {
        this.orderService = orderService;
    }

    /**
     * 創建訂單的API端點
     * @param order 訂單對象
     * @return 成功消息
     */
    @PostMapping
    public String createOrder(@RequestBody Order order) {
        orderService.createOrder(order);
        return"Order created successfully";
    }
}

5.訂單服務

/**
 * 訂單服務類，負責創建訂單並將訂單信息發送到相應的消息通道
 */
@Service
public class OrderService {

    private final OrderGateway gateway;

    @Autowired
    public OrderService(OrderGateway gateway) {
        this.gateway = gateway;
    }

    /**
     * 創建訂單並觸發訂單創建流程
     * @param order 訂單對象
     */
    public void createOrder(Order order) {
        System.out.println("Creating order: " + order.getOrderId());
        // 將訂單發送到orderCreatedChannel消息通道
        gateway.processOrder(order);
    }
}

6.支付處理服務

/**
 * 支付處理服務類，監聽訂單創建消息通道，處理支付邏輯
 */
@Component
public class PaymentService {

    private final OrderGateway gateway;

    @Autowired
    public PaymentService(OrderGateway gateway) {
        this.gateway = gateway;
    }

    /**
     * 處理訂單創建消息，模擬支付處理
     * @param order 訂單對象
     */
    @ServiceActivator(inputChannel = "orderCreatedChannel")
    public void handleOrderCreation(@Payload Order order) {
        System.out.println("Handling order creation for: " + order.getOrderId());
        // 模擬支付處理
        System.out.println("Processing payment for order: " + order.getOrderId());
        // 假設支付成功
        gateway.processPayment(order);
    }
}

7.庫存檢查服務

/**
 * 庫存檢查服務類，監聽支付處理消息通道，檢查庫存並決定是否發貨
 */
@Component
public class InventoryService {

    private final OrderGateway gateway;

    @Autowired
    public InventoryService(OrderGateway gateway) {
        this.gateway = gateway;
    }

    /**
     * 處理支付處理消息，檢查庫存
     * @param order 訂單對象
     */
    @ServiceActivator(inputChannel = "paymentProcessedChannel")
    public void checkInventory(@Payload Order order) {
        System.out.println("Checking inventory for product: " + order.getProductId());
        // 模擬庫存檢查
        boolean isInStock = true; // 假設庫存充足
        if (isInStock) {
            System.out.println("Product is in stock.");
            gateway.scheduleShipment(order);
        } else {
            System.out.println("Product is out of stock.");
            // 通知用户的邏輯,自己寫吧
        }
    }
}

8.發貨調度服務

/**
 * 發貨調度服務類，監聽發貨調度消息通道，安排發貨
 */
@Component
public class ShipmentService {

    /**
     * 處理髮貨調度消息，模擬發貨
     * @param order 訂單對象
     */
    @ServiceActivator(inputChannel = "shipmentScheduledChannel")
    public void scheduleShipment(@Payload Order order) {
        System.out.println("Scheduling shipment for order: " + order.getOrderId());
        // 模擬發貨調度
        System.out.println("Shipment scheduled for order: " + order.getOrderId());
    }
}

9.訂單處理相關的消息網關接口

/**
 * 定義訂單處理相關的消息網關接口
 */
public interface OrderGateway {

    /**
     * 將訂單發送到orderCreatedChannel消息通道
     * @param order 訂單對象
     */
    @Gateway(requestChannel = "orderCreatedChannel")
    void processOrder(Order order);

    /**
     * 將訂單發送到paymentProcessedChannel消息通道
     * @param order 訂單對象
     */
    @Gateway(requestChannel = "paymentProcessedChannel")
    void processPayment(Order order);

    /**
     * 將訂單發送到inventoryCheckedChannel消息通道
     * @param order 訂單對象
     */
    @Gateway(requestChannel = "inventoryCheckedChannel")
    void checkInventory(Order order);

    /**
     * 將訂單發送到shipmentScheduledChannel消息通道
     * @param order 訂單對象
     */
    @Gateway(requestChannel = "shipmentScheduledChannel")
    void scheduleShipment(Order order);
}

10.測試

curl -X POST http://localhost:8080/orders \
     -H "Content-Type: application/json" \
     -d '{"orderId": "123", "productId": "P001", "quantity": 2}'

11.測試日誌

Creating order: 123
Handling order creation for: 123
Processing payment for order: 123
Checking inventory for product: P001
Product is in stock.
Scheduling shipment for order: 123
Shipment scheduled for order: 123
springcloud , spring , JAVA , springboot
user avatar mannayang 頭像 king_wenzhinan 頭像 u_16297326 頭像 jianxiangjie3rkv9 頭像 sofastack 頭像 u_15702012 頭像 jiangyi 頭像 yizhidanshendetielian 頭像 lu_lu 頭像 gvison 頭像 lvweifu 頭像 chuck1sn 頭像
點贊 59 用戶, 點贊了這篇動態!
點贊

Add a new 評論

Some HTML is okay.