SpringBoot與Stateful Functions整合，實現實時用戶行為流處理與個性化推薦功能

作者：Java知識日歷 2025-08-08 02:11:00

Apache Flink Stateful Functions是一個輕量級、可擴展的狀態管理框架，旨在簡化復雜事件驅動系統的開發，可以通過定義和組合狀態化的函數來處理實時數據流。

哪些公司使用了Flink Stateful Functions？

Zalando 在其物流和供應鏈管理系統中使用 Flink Stateful Functions 來處理訂單跟蹤和庫存管理。
Netflix 使用 Flink Stateful Functions 來處理大規模的視頻流數據，包括推薦系統、內容分發網絡（CDN）優化等。
Capital One 利用 Flink Stateful Functions 進行實時信用評分和風險管理，確保貸款審批過程的高效和準確。
Airbnb 使用 Flink Stateful Functions 來處理用戶行為數據，優化住宿推薦和價格策略。
LinkedIn 使用 Flink Stateful Functions 來處理社交網絡中的各種事件流，如消息傳遞、通知推送等。
eBay 使用 Flink Stateful Functions 來處理廣告點擊流數據，優化廣告投放策略。

我們為什么選擇Stateful Functions？

Stateful Functions 提供了內置的狀態管理和容錯機制。每個函數實例可以擁有自己的狀態，這些狀態可以在故障恢復時自動重新加載，確保系統的穩定性和一致性。
通過模塊化的設計，Stateful Functions 可以方便地添加新的功能和業務邏輯，而無需重構整個系統。
Stateful Functions 提供了一種簡單且直觀的編程模型。我們只需關注具體的業務邏輯，而不需要處理底層的并發控制、狀態管理和網絡通信等復雜問題。
Stateful Functions 允許你定義多個相互關聯的函數來處理不同的事件類型，從而實現復雜的業務邏輯。

代碼實操

<dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-statefun-sdk-java8_2.12</artifactId>
        <version>3.0.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.flink</groupId>
        <artifactId>flink-clients_2.12</artifactId>
        <version>1.14.6</version>
    </dependency>
</dependencies>

<repositories>
    <repository>
        <id>apache.snapshots</id>
        <name>Apache Snapshot Repository</name>
        <url>https://repository.apache.org/content/repositories/snapshots/</url>
        <releases>
            <enabled>false</enabled>
        </releases>
        <snapshots>
            <enabled>true</enabled>
        </snapshots>
    </repository>
</repositories>

配置Flink Stateful Functions模塊

package com.example.demo.config;

import org.apache.flink.statefun.sdk.java.StatefulFunctions;
import org.apache.flink.statefun.sdk.java.TypeName;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
class StateFunConfig {

    // 定義Stateful Function的類型名稱
    publicstaticfinal TypeName RECOMMENDER_TYPE = TypeName.typeNameOf("com.example", "recommender");

    /**
     * 創建Stateful Functions實例
     * @return StatefulFunctions對象
     */
    @Bean
    public StatefulFunctions statefulFunctions() {
        return StatefulFunctions.builder()
                .withModule(new MyModule())
                .build();
    }

    /**
     * 自定義Flink Module類，用于配置Stateful Function
     */
    privatestaticclass MyModule implements org.apache.flink.statefun.sdk.java.Module {
        @Override
        public void configure(org.apache.flink.statefun.sdk.java.ModuleSpec spec) {
            // 注冊RecommenderFunction到模塊中
            spec.withStatefulFunction(RECOMMENDER_TYPE, new RecommenderFunction());
        }
    }
}

實現個性化推薦的Stateful Function

package com.example.demo.functions;

import org.apache.flink.statefun.sdk.Context;
import org.apache.flink.statefun.sdk.FunctionType;
import org.apache.flink.statefun.sdk.StatefulFunction;
import org.apache.flink.statefun.sdk.annotations.Persisted;
import org.apache.flink.statefun.sdk.io.EgressIdentifier;
import org.apache.flink.types.StringValue;

publicclass RecommenderFunction implements StatefulFunction {

    // 定義Egress標識符，用于將結果發送出去
    publicstaticfinal EgressIdentifier<String> RECOMMENDATION_EGRESS =
            new EgressIdentifier<>("com.example", "recommendations", String.class);

    // 持久化狀態變量，用于存儲推薦結果
    @Persisted
    privatefinal StringValue recommendations = new StringValue();

    /**
     * 處理輸入消息的方法
     * @param context 上下文信息
     * @param input 輸入的消息
     */
    @Override
    public void invoke(Context context, Object input) {
        // 假設輸入是一個用戶ID字符串
        String userId = (String) input;
        // 生成推薦結果
        String recommendation = generateRecommendation(userId);
        // 將推薦結果通過Egress發送出去
        context.send(RECOMMENDATION_EGRESS, userId, recommendation);
    }

    /**
     * 模擬生成推薦結果的方法
     * @param userId 用戶ID
     * @return 推薦結果字符串
     */
    private String generateRecommendation(String userId) {
        // 這里可以添加更復雜的業務邏輯來生成推薦結果
        return"Recommended products for user " + userId;
    }
}

Controller

package com.example.demo.controller;

import org.apache.flink.statefun.client.FlinkClient;
import org.apache.flink.statefun.sdk.Address;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;

@RestController
class UserController {

    privatefinal FlinkClient client;

    /**
     * 構造函數注入Flink客戶端
     * @param client Flink客戶端實例
     */
    @Autowired
    public UserController(FlinkClient client) {
        this.client = client;
    }

    /**
     * 處理POST請求，接收用戶ID并觸發Flink Stateful Function
     * @param userId 用戶ID
     * @return 處理結果字符串
     */
    @PostMapping("/user")
    public String processUser(@RequestBody String userId) {
        // 根據用戶ID創建Address對象
        Address address = Address.fromTypeNameAndId(com.example.demo.config.StateFunConfig.RECOMMENDER_TYPE, userId);
        // 發送用戶ID給Flink Stateful Function處理
        client.send(address, userId);
        // 返回處理結果
        return"Processing request for user: " + userId;
    }
}

Application

package com.example.demo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DemoApplication {

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

測試API

注意：確保你的Flink集群已經啟動并且可以通過Flink Client連接！！！切記！

curl -X POST http://localhost:8080/user -d 'user123'

Respons

Processing request for user: user123

責任編輯：武曉燕來源： Java知識日歷