.NET分散式Orleans - 7 - Streaming

来源:https://www.cnblogs.com/chenyishi/p/18103492
-Advertisement-
Play Games

概念 在Orleans中,Streaming是一組API和功能集,它提供了一種構建、發佈和消費數據流的方式。 這些流可以是任何類型的數據,從簡單的消息到複雜的事件或數據記錄。Streaming API允許你定義、發佈和消費這些流,而無需關心底層的傳輸機制或數據存儲。 每個流都有一個唯一的標識符,稱為 ...


概念

在Orleans中,Streaming是一組API和功能集,它提供了一種構建、發佈和消費數據流的方式。

這些流可以是任何類型的數據,從簡單的消息到複雜的事件或數據記錄。Streaming API允許你定義、發佈和消費這些流,而無需關心底層的傳輸機制或數據存儲。

每個流都有一個唯一的標識符,稱為StreamId,用於區分不同的流。流可以是持久的,也可以是臨時的,具體取決於所使用的流提供者(Stream Provider)。流提供者負責處理流的存儲、傳輸和故障恢復。

作用

Streaming在Orleans中起到了至關重要的作用,主要體現在以下幾個方面:

  1. 解耦:Streaming允許將數據的產生者和消費者解耦。生產者可以發佈數據到流中,而消費者可以獨立地訂閱這些流並處理數據。這種解耦使得系統更加靈活和可擴展。

  2. 實時性:通過Streaming,你可以實時地處理和響應數據流。這對於需要實時分析、監控或響應的場景非常有用。

  3. 故障恢復:Orleans的Streaming機制具有強大的故障恢復能力。即使在出現網路分區或節點故障的情況下,流提供者也能夠確保數據的可靠性和一致性。

應用場景

  1. 實時日誌分析:你可以將應用程式的日誌消息發佈到流中,並使用專門的消費者來分析這些日誌。這允許你實時地監控和響應應用程式的行為。

  2. 事件驅動架構:在事件驅動架構中,你可以使用Streaming來發佈事件,並由多個消費者來處理這些事件。這有助於構建松耦合、可擴展和響應式的系統。

  3. 分散式協作:Streaming也可以用於實現分散式系統中的協作和通信。例如,多個節點可以發佈狀態更新到流中,其他節點可以訂閱這些流以獲取最新的狀態信息。

示例

安裝nuget包

<PackageReference Include="Microsoft.Orleans.Streaming" Version="8.0.0" />

配置Streaming

siloHostBuilder.AddMemoryStreams("StreamProvider").AddMemoryGrainStorage("PubSubStore");

定義一個Grain生成事件

public interface ISender : IGrainWithStringKey
{
    Task Send(Guid rid);
}

public class SenderGrain : Grain, ISender
{
    public Task Send(Guid rid)
    {
        var streamProvider = this.GetStreamProvider("StreamProvider");
        var streamId = StreamId.Create("RANDOMDATA", rid);
        var stream = streamProvider.GetStream<int>(streamId);
        RegisterTimer(_ =>
        {
            return stream.OnNextAsync(Random.Shared.Next());
        }, null, TimeSpan.FromMilliseconds(1_000), TimeSpan.FromMilliseconds(1_000));
        return Task.CompletedTask;
    }
}

再定義一個Grain訂閱事件

public interface IRandomReceiver : IGrainWithGuidKey
{
    Task Receive();
}

[ImplicitStreamSubscription("RANDOMDATA")]
public class ReceiverGrain : Grain, IRandomReceiver
{
    public override async Task OnActivateAsync(CancellationToken cancellationToken)
    {
        var streamProvider = this.GetStreamProvider("StreamProvider");
        var rid = this.GetPrimaryKey();
        var streamId = StreamId.Create("RANDOMDATA", rid);
        var stream = streamProvider.GetStream<int>(streamId);

        await stream.SubscribeAsync<int>(
            async (data, token) =>
            {
                Console.WriteLine(data);
                await Task.CompletedTask;
            });
        base.OnActivateAsync(cancellationToken);
    }
    public async Task Receive()
    {
            
    }
}

然後即可測試

var rid = Guid.NewGuid();
var sender1 = client.GetGrain<ISender>("sender1");
await sender1.Send(rid);
var reciver1 = client.GetGrain<IRandomReceiver>(new Guid());
await reciver1.Receive();

流提供程式

提供程式可以通過在nuget種搜索Orleans.Streaming,也可以通過PersistentStreamProvider 與 IQueueAdapter 重寫來自定義Provider

 


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • socket在接收數據時,經常會因為網路延遲、緩存區數據處理不及時等原因造成收到的數據是多個包黏在一起的情況。如下圖所示 圖中紅色框部分是通訊的 心跳包 圖中黃色框部分和未框選部分是 兩包 數據包 所以可見此時緩存區裡面同時存在了一個心跳包,兩個數據包 如何分包(此處僅針對幾個完整的數據包在一起的情 ...
  • 首先看完成效果 一個玩家的效果 多個玩家的效果 源碼地址 https://gitee.com/chesterdotchen/snake-with-orleans 項目介紹 Snake.Common項目 IGameGrain:游戲的Grain定義,與State定義 ISnakeGrain:蛇的Grai ...
  • 本章將和大家分享 Elasticsearch 中的數據聚合功能,通過聚合(aggregations)可以實現對文檔數據的統計、分析、運算。 ...
  • 工具類擴展 1. ILSpy 2022 (免費) ILSpy 是 ILSpy 開源反編譯器的 Visual Studio 擴展。 是一款開源、免費的、且適用於.NET平臺反編譯【C#語言編寫的程式和庫(.dll)內容】工具;可以集成在Visual Studio 開發工具中,能夠十分快捷方便的查看源代 ...
  • 前言 WPF中Window相信大家都很熟悉,有時我們有一些自定義需求預設Window是無法滿足的,比如在標題欄上放一些自己東西,這個時候我們就需要寫一個自己的Window,實現起來也很簡單,只要給Window設置一個WindowChrome.WindowChrome附加屬性就可以實現,WindowC ...
  • 前言 上文介紹瞭如何通過一個Form自定義控制項來簡化數據的錄入,並自動實現數據校驗,自動佈局排列等功能。本文繼續介紹如何優化表格控制項的使用,縮減代碼量,實現工作效率的提升。 一、功能實現 上文中分析了DataGrid跟ListView兩種表格控制項的優劣,在這裡我們選擇ListView來實現我們的表格 ...
  • gRPC是一個現代的開源高性能遠程過程調用(RPC)框架,它可以高效地連接數據中心內和跨數據中心的服務,支持負載平衡、跟蹤、運行狀況檢查和身份驗證。 gRPC通過使用 Protocol Buffers 作為數據傳輸格式,實現了在不同平臺上的通信,並支持雙向流和流式傳輸。RPC 是遠程過程調用的縮寫, ...
  • 到目前為止,Orleans7的核心概念基本已經學完,我準備使用Orleans7做一個項目實戰,來總結自己的學習效果。 項目效果 通過Orleans7來完成一個貪吃蛇游戲,要求如下: 可以多人線上玩 貪吃蛇可以上/下/左/右改方向 貪吃蛇吃完食物,身體長度+1 項目暫定架構 初步設想,此游戲包括一個O ...
一周排行
    -Advertisement-
    Play Games
  • 隨著Aspire發佈preview5的發佈,Microsoft.Extensions.ServiceDiscovery隨之更新, 服務註冊發現這個屬於老掉牙的話題解決什麼問題就不贅述了,這裡主要講講Microsoft.Extensions.ServiceDiscovery(preview5)以及如何 ...
  • 概述:通過使用`SemaphoreSlim`,可以簡單而有效地限制非同步HTTP請求的併發量,確保在任何給定時間內不超過20個網頁同時下載。`ParallelOptions`不適用於非同步操作,但可考慮使用`Parallel.ForEach`,儘管在非同步場景中謹慎使用。 對於併發非同步 I/O 操作的數量 ...
  • 1.Linux上安裝Docken 伺服器系統版本以及內核版本:cat /etc/redhat-release 查看伺服器內核版本:uname -r 安裝依賴包:yum install -y yum-utils device-mapper-persistent-data lvm2 設置阿裡雲鏡像源:y ...
  • 概述:WPF界面綁定和渲染大量數據可能導致性能問題。通過啟用UI虛擬化、非同步載入和數據分頁,可以有效提高界面響應性能。以下是簡單示例演示這些優化方法。 在WPF中,當你嘗試綁定和渲染大量的數據項時,性能問題可能出現。以下是一些可能導致性能慢的原因以及優化方法: UI 虛擬化: WPF提供了虛擬化技術 ...
  • 引言 上一章節介紹了 TDD 的三大法則,今天我們講一下在單元測試中模擬對象的使用。 Fake Fake - Fake 是一個通用術語,可用於描述 stub或 mock 對象。 它是 stub 還是 mock 取決於使用它的上下文。 也就是說,Fake 可以是 stub 或 mock Mock - ...
  • 為.net6在CentOS7上面做準備,先在vmware虛擬機安裝CentOS 7.9 新建CentOS764位的系統 因為CentOS8不更新了,所以安裝7;簡單就一筆帶過了 選擇下載好的操作系統的iso文件,下載地址https://mirrors.aliyun.com/centos/7.9.20 ...
  • 經過前面幾篇的學習,我們瞭解到指令的大概分類,如:參數載入指令,該載入指令以 Ld 開頭,將參數載入到棧中,以便於後續執行操作命令。參數存儲指令,其指令以 St 開頭,將棧中的數據,存儲到指定的變數中,以方便後續使用。創建實例指令,其指令以 New 開頭,用於在運行時動態生成並初始化對象。方法調用指... ...
  • LiteDB 是一個輕量級的嵌入式 NoSQL 資料庫,其設計理念與 MongoDB 類似,但它是完全使用 C# 開發的,因此與 C# 應用程式的集成非常順暢。與 SQLite 相比,LiteDB 提供了 NoSQL(即鍵值對)的數據存儲方式,並且是一個開源且免費的項目。它適用於桌面、移動以及 We ...
  • 1 開源解析和拆分文檔 第三方的工具去對文件解析拆分,去將我們的文件內容給提取出來,並將我們的文檔內容去拆分成一個小的chunk。常見的PDF word mark down, JSON、HTML。都可以有很好的一些模塊去把這些文件去進行一個東西去提取。 優勢 支持豐富的文檔類型 每種文檔多樣化選擇 ...
  • OOM是什麼?英文全稱為 OutOfMemoryError(記憶體溢出錯誤)。當程式發生OOM時,如何去定位導致異常的代碼還是挺麻煩的。 要檢查OOM發生的原因,首先需要瞭解各種OOM情況下會報的異常信息。這樣能縮小排查範圍,再結合異常堆棧、heapDump文件、JVM分析工具和業務代碼來判斷具體是哪 ...