使用?Rust?實(shí)現(xiàn)的基礎(chǔ)的List?和?Watch?機(jī)制示例流程

更新時(shí)間：2025年09月23日 08:47:50 作者：天翼云開發(fā)者社區(qū)

本文給大家介紹使用Rust實(shí)現(xiàn)的基礎(chǔ)的List和Watch機(jī)制示例流程,本文通過(guò)實(shí)例代碼給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友參考下吧

使用 Rust 實(shí)現(xiàn)的基礎(chǔ)的List 和 Watch 機(jī)制

介紹

在日常的開發(fā)過(guò)程中，有一個(gè)很重要的任務(wù)是能夠通過(guò)Rust語(yǔ)言實(shí)現(xiàn)K8s中的各種生態(tài)組件，在這個(gè)過(guò)程中，既需要能過(guò)夠了解K8S的工作原理也需要能夠知道rust的語(yǔ)言特性。因此，在這個(gè)過(guò)程中有很多值得探討的知識(shí)點(diǎn)。

在這里，第一步，我們將探索如何使用 Rust 實(shí)現(xiàn)一個(gè)類似于 Kubernetes 的 list 和 watch 機(jī)制。我們將通過(guò) WebSocket 實(shí)現(xiàn)實(shí)時(shí)的消息推送，并使用一些關(guān)鍵的 Rust 異步編程模型來(lái)處理事件和連接管理。

我們首先默認(rèn)大家能夠了解rust語(yǔ)言的基本特性。下文中，將針對(duì)rust的知識(shí)點(diǎn)展開進(jìn)行探討。

目標(biāo)

理解 WebSocket 連接的建立和管理。
學(xué)習(xí)如何通過(guò) WebSocket 推送消息。
掌握消息緩存和處理的實(shí)現(xiàn)方式。
了解如何使用 Rust 實(shí)現(xiàn)一個(gè)高效的事件分發(fā)系統(tǒng)。
理解K8S中的數(shù)據(jù)一致性保障方法
了解本機(jī)制的不足，以及后續(xù)如何進(jìn)行改進(jìn)

理解問(wèn)題

什么是 list 和 watch？

List：列出當(dāng)前所有資源的狀態(tài)。
Watch：實(shí)時(shí)監(jiān)控資源的變化，一旦有資源變化，就會(huì)立即通知客戶端。

使用場(chǎng)景

自動(dòng)化運(yùn)維：實(shí)時(shí)監(jiān)控系統(tǒng)資源狀態(tài)，觸發(fā)自動(dòng)化運(yùn)維操作。
應(yīng)用監(jiān)控：實(shí)時(shí)獲取應(yīng)用狀態(tài)，及時(shí)處理異常，在很多的系統(tǒng)設(shè)計(jì)場(chǎng)景中，能夠減少耦合。
K8S中的相應(yīng)設(shè)計(jì)：K8S中，對(duì)相應(yīng)資源的通知的基礎(chǔ)即為list and watch機(jī)制。本人在學(xué)習(xí)K8S源碼的第一步就是學(xué)習(xí)這一套設(shè)計(jì)架構(gòu)。

分析問(wèn)題

\當(dāng)然，通過(guò)簡(jiǎn)單的代碼僅僅通過(guò)http進(jìn)行主動(dòng)連接也可實(shí)現(xiàn)這個(gè)功能。但在目前階段，我們希望能夠設(shè)計(jì)一個(gè)高效的、穩(wěn)定的、可擴(kuò)展的list and watch體系，因此我們需要考慮以下幾個(gè)關(guān)鍵問(wèn)題。

關(guān)鍵問(wèn)題

如何建立和管理我們服務(wù)器和客戶端的連接？通過(guò)什么方式進(jìn)行？
如何實(shí)現(xiàn)高效的消息推送機(jī)制？
如何處理消息緩存和訂閱管理？

技術(shù)選型

語(yǔ)言：Rust
Web 框架：warp框架
WebSocket實(shí)現(xiàn)和框架：tokio-tungstenite、warp
異步編程：tokio、管道機(jī)制

設(shè)計(jì)代碼結(jié)構(gòu)

針對(duì)以上這個(gè)需求，結(jié)合目前kunos-system的需求我們闡釋如下

有以下幾個(gè)資源，Node、Task（Task是一個(gè)shell命令、鏡像運(yùn)行命令的載體）、Job（Task的上層資源，一個(gè)Job包含多個(gè)Task，類似于K8s中的replicaset）我們需要對(duì)這幾個(gè)資源的狀態(tài)進(jìn)行推送。
能夠在服務(wù)器建立起來(lái)一個(gè)watch and list服務(wù)器，能夠推送各種事件
能夠

組件設(shè)計(jì)

Broker：管理 WebSocket 訂閱者和事件分發(fā)。

pub struct Broker<R: Resource + Clone + Serialize + Send + Sync  + 'static> {
    // 下游的訂閱者列表，用于發(fā)送websocket信息
    subscribers: Arc<RwLock<HashMap<Topic, HashMap<Uuid, WsSender>>>>,
    // 事件的緩沖流
    event_sender: UnboundedSender<(Topic, WatchEvent<R>)>,
}

Watcher：對(duì)不同資源類型進(jìn)行管理和操作。

pub struct Watcher {
    // 為不同的事件建立不同的broker
    pub node_broker: Arc<Broker<Node>>,
    pub task_broker: Arc<Broker<Task>>,
    pub job_broker: Arc<Broker<Job>>,
    pub exec_broker: Arc<Broker<TaskExecRequest>>,
}

WebSocket 客戶端：與服務(wù)器交互，接收實(shí)時(shí)事件。

基本原理

websocket路由入口

let node_subscribe = warp::path!("watch" / "node").and(warp::ws()).map(
    move |ws: warp::ws::Ws| {
        let node_broker_clone = Arc::clone(&node_broker_clone);
        ws.on_upgrade(move |socket| async move {
            node_broker_clone.subscribe("node".to_string(), socket).await;
        })
    },
);

1. warp::path!("watch" / "node")

*這部分代碼定義了一個(gè)路徑過(guò)濾器，用于匹配路徑 /watch/node 的 HTTP 請(qǐng)求。warp::path!是 Warp 框架提供的一個(gè)宏，用于簡(jiǎn)化路徑定義。這里的"watch" / "node"表示請(qǐng)求路徑必須是/watch/node` 才能匹配這個(gè)過(guò)濾器。

2. .and(warp::ws())

這一部分代碼將路徑過(guò)濾器與 WebSocket 協(xié)議過(guò)濾器組合起來(lái)。warp::ws() 過(guò)濾器會(huì)匹配 WebSocket 握手請(qǐng)求并提取一個(gè) warp::ws::Ws 類型，表示 WebSocket 配置。這表示我們的這個(gè)路徑將為一個(gè)websocket接口。

warp::ws() 過(guò)濾器用于匹配并提取 WebSocket 握手請(qǐng)求，確保該請(qǐng)求是 WebSocket 協(xié)議請(qǐng)求。

3. .map(move |ws: warp::ws::Ws| { ... })

.map 方法用于將前面的過(guò)濾器組合結(jié)果映射到一個(gè)新的處理邏輯中。這里的 move |ws: warp::ws::Ws| { ... } 是一個(gè)閉包，用于處理 WebSocket 請(qǐng)求。

move 關(guān)鍵字確保閉包捕獲其環(huán)境中的所有變量的所有權(quán)，因?yàn)檫@些變量將在異步操作中使用。
ws: warp::ws::Ws 參數(shù)是從前面的 warp::ws() 過(guò)濾器中提取的 WebSocket 配置。

4. ws.on_upgrade(move |socket| async move { ... })

ws.on_upgrade 方法用于將 WebSocket 協(xié)議升級(jí)請(qǐng)求處理為 WebSocket 連接。它接受一個(gè)閉包作為參數(shù)，當(dāng) WebSocket 握手成功后，這個(gè)閉包會(huì)被調(diào)用。在官方定義中，這個(gè)方法主要用于自定義一個(gè)函數(shù)對(duì)建立后的websocket連接進(jìn)行一定的操作，因此我們?cè)谶@里將建立連接后一切操作，比如保持連接，發(fā)送信息等。

/// Finish the upgrade, passing a function to handle the `WebSocket`.
///
/// The passed function must return a `Future`.
pub fn on_upgrade<F, U>(self, func: F) -> impl Reply
where
    F: FnOnce(WebSocket) -> U + Send + 'static,
    U: Future<Output = ()> + Send + 'static,
{
    WsReply {
        ws: self,
        on_upgrade: func,
    }
}

move |socket| async move { ... } 是一個(gè)異步閉包，它將在 WebSocket 連接成功升級(jí)后執(zhí)行。
socket 參數(shù)表示已經(jīng)升級(jí)的 WebSocket 連接。

5. node_broker_clone.subscribe("node".to_string(), socket).await;

在異步閉包內(nèi)部，調(diào)用 node_broker_clone 的subscribe` 方法，將新的 WebSocket 連接訂閱到節(jié)點(diǎn)（node）主題中。后續(xù)我們將展開講解

"node".to_string() 將節(jié)點(diǎn)主題名稱轉(zhuǎn)換為字符串。
socket 參數(shù)表示當(dāng)前的 WebSocket 連接。
await 關(guān)鍵字等待異步訂閱操作完成。

websocket連接處理

上面說(shuō)到，我們通過(guò) ws.on_upgrade(move |socket| async move { ... })這個(gè)方法在連接建立之后進(jìn)行處理，其中可以知道，我們處理的方法如下所示。

pub async fn subscribe(&self, topic: Topic, socket: warp::ws::WebSocket) {
        let (ws_sender, mut ws_receiver) = socket.split();
        let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
        let subscriber_id = Uuid::new_v4();
?
        {
            let mut subs = self.subscribers.write().await;
            subs.entry(topic.clone()).or_default().insert(subscriber_id, tx);
        }
?
        let subscribers = Arc::clone(&self.subscribers);
        tokio::task::spawn(async move {
            while let Some(result) = ws_receiver.next().await {
                match result {
                    Ok(message) => {
                        // 處理有效的消息
                        if message.is_text() {
                            println!(
                                "Received message from client: {}",
                                message.to_str().unwrap()
                            );
                        }
                    }
                    Err(e) => {
                        // 處理錯(cuò)誤
                        eprintln!("WebSocket error: {:?}", e);
                        break;
                    }
                }
            }
            println!("WebSocket connection closed");
            subscribers.write().await.get_mut(&topic).map(|subscribers| subscribers.remove(&subscriber_id));
        });
?
        tokio::task::spawn(async move {
            let mut sender = ws_sender;
?
            while let Some(msg) = rx.recv().await {
                let _ = sender.send(msg).await;
            }
        });
    }

websocket連接處理 let (ws_sender, mut ws_receiver) = socket.split();這里使用原生的代碼，將已經(jīng)建立起來(lái)的socket進(jìn)行分割，因?yàn)閣ebsocket是雙向連接，因此獲得針對(duì)這個(gè)socket的發(fā)送端（ws_sender）和接收端(ws_receiver)。

建立連接并保存

let (tx, mut rx) = mpsc::unbounded_channel::<Message>();
let subscriber_id = Uuid::new_v4();
?
{
    let mut subs = self.subscribers.write().await;
    subs.entry(topic.clone()).or_default().insert(subscriber_id, tx);
}

在這里，我們建立了個(gè)一個(gè)管道，并將subscriber的信息進(jìn)行保存，這里的 mpsc::unbounded_channel::<Message>();類似于golang中的channel，他會(huì)生成一個(gè)發(fā)送者、一個(gè)接收者，當(dāng)往發(fā)送者發(fā)送消息的時(shí)候，接收者會(huì)受到該消息并進(jìn)行一定處理。因此我們將subscriber的發(fā)送者（tx）保存至內(nèi)存里。

建立消息發(fā)送機(jī)制

tokio::task::spawn(async move {
            let mut sender = ws_sender;
?
            while let Some(msg) = rx.recv().await {
                let _ = sender.send(msg).await;
            }
        });

這個(gè)就是很簡(jiǎn)單了，通過(guò)如果rx收到了消息，則向websocket的subscriber進(jìn)行發(fā)送。該任務(wù)是以新協(xié)程任務(wù)的方式啟動(dòng)的，在后臺(tái)持續(xù)運(yùn)行

建立websocket連接?；顧C(jī)制

let subscribers = Arc::clone(&self.subscribers);
    tokio::task::spawn(async move {
        while let Some(result) = ws_receiver.next().await {
            match result {
                Ok(message) => {
                    // 處理有效的消息
                    if message.is_text() {
                        println!(
                            "Received message from client: {}",
                            message.to_str().unwrap()
                        );
                    }
                }
                Err(e) => {
                    // 處理錯(cuò)誤
                    eprintln!("WebSocket error: {:?}", e);
                    break;
                }
            }
        }
        println!("WebSocket connection closed");
        subscribers.write().await.get_mut(&topic).map(|subscribers| subscribers.remove(&subscriber_id));
    });

這里我們?nèi)匀辉诤笈_(tái)啟動(dòng)一個(gè)守護(hù)協(xié)程，用于?；顆ebsocket連接，一旦發(fā)生了連接失效，則注銷消息發(fā)送機(jī)制，刪除subscribers緩存中的訂閱者。

消息推送機(jī)制

事件推送事件推送時(shí)候?qū)⒃试S調(diào)用相關(guān)事件的推送地址，向推送端發(fā)送消息。

pub async fn produce_node_event(&self, event: WatchEvent<Node>) {
        self.node_broker.produce("node".to_string(), event).await;
    }
    pub async fn produce_task_event(&self, event: WatchEvent<Task>) {
        self.task_broker.produce("task".to_string(), event).await;
    }
    pub async fn produce_job_event(&self, event: WatchEvent<Job>) {
        self.job_broker.produce("job".to_string(), event).await;
    }

當(dāng)收到消息的時(shí)候，不直接處理消息，而是將放入緩存隊(duì)列中（一個(gè)消息無(wú)界流）

pub async fn produce(&self, topic: Topic, event: WatchEvent<R>) {
        if let Err(e) = self.event_sender.send((topic.clone(), event.clone())) {
            eprintln!("Failed to send event: {}", e);
        }
    }

事件分發(fā)同樣的。將啟動(dòng)一個(gè)協(xié)程，用于從和event_sender對(duì)應(yīng)的event_receiver中獲取消息，推送給訂閱者。

fn start_event_dispatcher(broker: Arc<Self>, mut event_receiver: UnboundedReceiver<(Topic, WatchEvent<R>)>) {
        tokio::spawn(async move {
            while let Some((topic, event)) = event_receiver.recv().await {
                let event_json = serde_json::to_string(&event).unwrap();
                let subscribers_list;
                {
                    let subscribers = broker.subscribers.read().await;
                    subscribers_list = subscribers.get(&topic).cloned().unwrap_or_default();
                }
                let mut invalid_subscribers = vec![];
                for (id, ws_sender) in subscribers_list {
                    if ws_sender.send(warp::ws::Message::text(event_json.clone())).is_err() {
                        invalid_subscribers.push(id);
                    }
                }
                if !invalid_subscribers.is_empty() {
                    let mut subscribers = broker.subscribers.write().await;
                    if let Some(subscribers) = subscribers.get_mut(&topic) {
                        for id in invalid_subscribers {
                            subscribers.remove(&id);
                        }
                    }
                }
            }
        });
    }

獲取訂閱者的列表并依次發(fā)送

如果發(fā)現(xiàn)發(fā)送失敗，則將這個(gè)訂閱者從緩存中刪除

客戶端

客戶端的代碼就是建立起來(lái)一個(gè)訂閱者關(guān)注相關(guān)事件的動(dòng)態(tài)。在相應(yīng)的代碼中，可以使用該方法。本方法最終返回的是一個(gè)無(wú)界流 Stream<Item = WatchEvent<R>>，用于得到服務(wù)器推送過(guò)來(lái)的事件類型

pub async fn list_and_watch<R>(api_client: &ApiClient, resource_name: &str) -> impl Stream<Item = WatchEvent<R>>
where
    R: Resource + Clone + DeserializeOwned + 'static + Send,
{
    // 先通過(guò) HTTP 獲取資源列表
    let initial_resources = get_resource_list::<R>(api_client).await;
    // 解析要連接WebSocket服務(wù)器的URL
    let url = Url::parse(&*format!("{}/{}", api_client.watch_url, resource_name)).expect("Invalid URL");
    // 連接到WebSocket服務(wù)器
    println!("watch url is {}", url);
    let (ws_stream, _) = connect_async(url).await.expect("Failed to connect");
    println!("Watch client connected");
    let (mut write, read) = ws_stream.split();
    let (tx, rx) = mpsc::unbounded_channel();
    // 先發(fā)送初始資源列表
    match initial_resources {
        Ok(res) => tx.send(WatchEvent::Restarted(res)).unwrap(),
        Err(e) => eprintln!("list resource failed, {}", e),
    };
    // 將 WebSocket 讀流轉(zhuǎn)換為消息事件流
    tokio::spawn(async move {
        read.for_each(|message| async {
            match message {
                Ok(msg) => {
                    if msg.is_text() {
                        let text = msg.to_text().unwrap();
                        match serde_json::from_str::<WatchEvent<R>>(text) {
                            Ok(event) => {
                                tx.send(event).unwrap();
                            }
                            Err(e) => {
                                eprintln!("Failed to parse message: {:?}", e);
                            }
                        }
                    }
                }
                Err(e) => {
                    eprintln!("Error receiving message: {:?}", e);
                }
            }
        }).await;
    });
    // 保持 WebSocket 連接活躍
    tokio::spawn(async move {
        loop {
            if let Err(e) = write.send(WatchMessage::Text(String::new())).await {
                eprintln!("Error sending ping: {:?}", e);
                break;
            }
            tokio::time::sleep(tokio::time::Duration::from_secs(10)).await;
        }
    });
    tokio_stream::wrappers::UnboundedReceiverStream::new(rx)
}

使用驗(yàn)證

不足分析

經(jīng)過(guò)上面的介紹，我們可以看到這個(gè)基礎(chǔ)的list and watch機(jī)制能夠正確運(yùn)行。但是，和K8S、ETCD中廣泛使用的list and watch相比仍然缺少一個(gè)機(jī)制來(lái)保證list和watch的一致性。

請(qǐng)考慮這樣一種情況我們的服務(wù)器中會(huì)源源不斷地產(chǎn)生數(shù)據(jù)d1,d2,d3,...,dn。當(dāng)我們使用list時(shí)候，能夠感知到d1,d2,d3，此時(shí)我們完成list，開始建立watch。加入在開始建立watch這個(gè)階段，即使可能是幾毫秒的時(shí)間但服務(wù)器生成了d4，而在watch建立起來(lái)后，只能接收到d5,d6,...。這就導(dǎo)致了數(shù)據(jù)的遺失。

在 Kubernetes 中，List 和 Watch 操作結(jié)合使用時(shí)，需要使用一個(gè)revision機(jī)制以確保資源的變更不會(huì)被遺漏。理解 List 和 Watch 操作時(shí) revision（即 resourceVersion）的具體含義和管理方式對(duì)于保證一致性至關(guān)重要。revision的存在有著如下的意義：

數(shù)據(jù)版本控制：revision 是 Etcd 的全局遞增計(jì)數(shù)器，用于標(biāo)識(shí)數(shù)據(jù)的當(dāng)前版本。當(dāng)進(jìn)行數(shù)據(jù)的修改、更新操作時(shí)候，revision會(huì)+1
一致性視圖：確保返回的數(shù)據(jù)是一致的快照視圖，表示在該 revision 之前的所有操作都已完成。

revision 與 List 和 Watch 的關(guān)系

List 操作：
- 返回資源列表和當(dāng)前的全局 revision，作為 resourceVersion。
- 確保獲取到的資源是該 revision 時(shí)刻的一致視圖。
Watch 操作：
- 使用 List 操作返回的resourceVersion` 作為起點(diǎn)。
- 從該 resourceVersion 開始監(jiān)聽資源的變化，確保在List 和Watch` 之間的變更不會(huì)丟失。

List 操作的 revision

當(dāng)進(jìn)行 List 操作時(shí)，Kubernetes API Server 從 Etcd 獲取當(dāng)前資源的狀態(tài)及其resourceVersion 。這個(gè) resourceVersion 是 Etcd 當(dāng)前的全局revision 。它表示在此 revision 之前的所有操作都已經(jīng)完成，并確保返回的數(shù)據(jù)是這個(gè)revision` 時(shí)刻的一致視圖。

Watch 操作的 revision

Watch 操作使用 List 操作返回的 resourceVersion 作為起點(diǎn)，從該版本開始監(jiān)聽資源的變化。這確保了從 List 到 Watch 之間的變更不會(huì)被遺漏。

示例流程

List 操作：
- API Server 從 Etcd 獲取指定資源的當(dāng)前狀態(tài)。
- Etcd 返回包含所有資源對(duì)象的列表和一個(gè)全局 revision ，這個(gè) revision 將作為resourceVersion`。
Watch 操作：
- API Server 使用 List 操作返回的 resourceVersion（revision）作為起點(diǎn)，開始監(jiān)聽資源的變化。
- Etcd 返回從指定 revision` 開始的所有變更事件。

總結(jié)

revision：標(biāo)識(shí)數(shù)據(jù)版本，確保數(shù)據(jù)一致性。
List 和 Watch：List 獲取資源和 revision，Watch 從該 revision 開始監(jiān)聽變化，確保變更的連續(xù)性和一致性。

到此這篇關(guān)于使用 Rust 實(shí)現(xiàn)的基礎(chǔ)的List 和 Watch 機(jī)制的文章就介紹到這了,更多相關(guān)Rust List 和 Watch 機(jī)制內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: