互操作

来源:https://www.cnblogs.com/BigBrotherStone/archive/2020/02/01/12247598.html
-Advertisement-
Play Games

" 返回《C 併發編程》" "1. 用 async 代碼封裝非同步方法與 Completed 事件" "2. 用 async 代碼封裝 Begin/End 方法" "3. 用 async 代碼封裝並行代碼" "4. 用 async 代碼封裝 Rx Observable 對象" "5. 用 Rx Obs ...


>>返回《C# 併發編程》

非同步封裝

1. 用 async 代碼封裝非同步方法與 Completed 事件

public static void MyDownloadStringTaskAsyncRun()
{
    WebClient client = new WebClient();
    string res = client.MyDownloadStringTaskAsync(new Uri("http://www.baidu.com")).Result;
    System.Console.WriteLine(res);
}

public static Task<string> MyDownloadStringTaskAsync(this WebClient client, Uri address)
{
    var tcs = new TaskCompletionSource<string>();
    // 這個事件處理程式會完成 Task 對象，並自行註銷。
    DownloadStringCompletedEventHandler handler = null;
    handler = (_, e) =>
    {
        client.DownloadStringCompleted -= handler;
        if (e.Cancelled)
            tcs.TrySetCanceled();
        else if (e.Error != null)
            tcs.TrySetException(e.Error);
        else
            tcs.TrySetResult(e.Result);
    };
    // 登記事件，然後開始操作。
    client.DownloadStringCompleted += handler;
    client.DownloadStringAsync(address);
    return tcs.Task;
}

輸出:

<!DOCTYPE html><!--STATUS OK-->
<html>
... ...
</html>

2. 用 async 代碼封裝 Begin/End 方法

public static void GetResponseAsyncRun()
{
    WebRequest request = WebRequest.Create("http://www.baidu.com");
    var response = request.MyGetResponseAsync().Result;

    System.Console.WriteLine($"WebResponse.ContentLength:{response.ContentLength}");
}
public static Task<WebResponse> MyGetResponseAsync(this WebRequest client)
{
    return Task<WebResponse>.Factory.FromAsync(client.BeginGetResponse, client.EndGetResponse, null);
}

輸出:

WebResponse.ContentLength:14615
  • 建議： 要在調用 FromAsync 之前調用 BeginOperation 。
  • 調用 FromAsync ，並讓用 BeginOperation 方法返回的 IAsyncOperation 作為參數，這樣也是可以的，但是 FromAsync 會採用效率較低的實現方式。

3. 用 async 代碼封裝並行代碼

await Task.Run(() => Parallel.ForEach(...));

通過使用 Task.Run ，所有的並行處理過程都推給了線程池。

Task.Run 返回一個代表並行任務的 Task 對象

  • UI 線程可以（非同步地）等待它完成（非阻塞）

4. 用 async 代碼封裝 Rx Observable 對象

事件流中幾種可能關註的情況：

  • 事件流結束前的最後一個事件；
  • 下一個事件；
  • 所有事件。
public delegate void HelloEventHandler(object sender, HelloEventArgs e);
public class HelloEventArgs : EventArgs
{
    public string Name { get; set; }
    public HelloEventArgs(string name)
    {
        Name = name;
    }
    public int SayHello()
    {
        System.Console.WriteLine(Name + " Hello.");
        return DateTime.Now.Millisecond;
    }
}

public static event HelloEventHandler HelloHandlerEvent;
public static void FirstLastRun()
{
    var task = Task.Run(() =>
    {
        Thread.Sleep(500);
        HelloHandlerEvent?.Invoke(new object(), new HelloEventArgs("lilei"));
        HelloHandlerEvent?.Invoke(new object(), new HelloEventArgs("HanMeimei"));
        HelloHandlerEvent?.Invoke(new object(), new HelloEventArgs("Tom"));
        HelloHandlerEvent?.Invoke(new object(), new HelloEventArgs("Jerry"));

    });

    var observable = Observable.FromEventPattern<HelloEventHandler, HelloEventArgs>(
     handler => (s, a) => handler.Invoke(s, a), handler => HelloHandlerEvent += handler, handler => HelloHandlerEvent -= handler)
     .Select(evt => evt.EventArgs.SayHello()).ObserveOn(Scheduler.Default)
     .Select(s =>
     {
        // 複雜的計算過程。
        Thread.Sleep(100);
        var result = s;
        Console.WriteLine("Now Millisecond result " + result + " on thread " + Environment.CurrentManagedThreadId);
        return result;
     })
     .Take(3)//這個標識3個就結束了
     ;

    var res =
        Task.Run(async () => await observable
    // //4個hello，3個result，res為最後一個的結果
    //.FirstAsync()//4個hello，1個result，res為第一個的結果
    //.LastAsync()//4個hello，3個result，res為最後一個的結果
    //.ToList()//4個hello，3個result，res為3個的結果
    ).Result;
    System.Console.WriteLine($"Res:{string.Join(',', res)},ResType:{res.GetType().Name}");

    task.Wait();
}

輸出:

lilei Hello.
HanMeimei Hello.
Tom Hello.
Jerry Hello.
Now Millisecond result 534 on thread 7
Now Millisecond result 544 on thread 7
Now Millisecond result 544 on thread 7
Res:544,ResType:Int32

在 await 調用 Observable 對象或 LastAsync 時，代碼（非同步地）等待事件流完成，然後返 回最後一個元素。

  • 在內部，await 實際是在訂閱事件流，完成後退訂
    cs IObservable<int> observable = ...; int lastElement = await observable.LastAsync(); // 或者 int lastElement = await observable;

使用 FirstAsync 可捕獲事件流中， FirstAsync 方法執行後的下一個事件。

  • 本例中 await 訂閱事件流，然後在第一個事件到達後立即結束（並退訂）：
    cs IObservable<int> observable = ...; int nextElement = await observable.FirstAsync();

使用 ToList 可捕獲事件流中的所有事件：

IObservable<int> observable = ...;     
IList<int> allElements = await observable.ToList();

5. 用 Rx Observable 對象封裝 async 代碼

任何非同步操作都可看作一個滿足以下條件之一的可觀察流：

  • 生成一個元素後就完成；
  • 發生錯誤，不生成任何元素。

ToObservable 和 StartAsync 都會立即啟動非同步操作，而不會等待訂閱

  • 但之後訂閱呢，或等待執行完再訂閱呢，能得到結果嗎
    • 可以，後面例子中的“輸出”中有體現

如果要讓 observable 對象在接受訂閱後才啟動操作，可使用 FromAsync

  • 跟 StartAsync 一樣，它也支持使用 CancellationToken取消
public static void AsyncObservableRun()
{
    var client = new HttpClient();

    IObservable<int> response1 = Task.Run(() => { System.Console.WriteLine("Run 1."); return 1; }).ToObservable();//直接執行

    IObservable<int> response2 = Observable.StartAsync(token => Task.Run(() => { System.Console.WriteLine("Run 2."); return 2; }, token));//直接執行

    IObservable<int> response3 = Observable.FromAsync(token => Task.Run(() => { System.Console.WriteLine("Run 3."); return 3; }, token));//訂閱後執行

    var res = Task.Run(async () =>
        await response1
        //await response2
        //await response3
    ).Result;
    System.Console.WriteLine($"Res:{res}");
}

輸出(response1):

Run 1.
Run 2.
Res:1

輸出(response2):

Run 1.
Run 2.
Res:2

輸出(response1):

Run 1.
Run 2.
Run 3.
Res:3
  • ToObservable 和 StartAsync 都返回一個 observable 對象，表示一個已經啟動的非同步操作
  • FromAsync 在每次被訂閱時都會啟動一個全新獨立的非同步操作。

下麵的例子使用一個已有的 URL 事件流，在每個 URL 到達時發出一個請求：

public static void SelectManyRun()
{
    IObservable<int> nums = new int[] { 1, 2, 3 }.ToObservable();
    IObservable<int> observable = nums.SelectMany((n, token) => Task.Run<int>(() => { System.Console.WriteLine($"Run {n}."); return n + 1; }, token));

    var res = Task.Run(async () => await observable.LastAsync()).Result;

    System.Console.WriteLine($"Res:{res}");
}

輸出:

Run 1.
Run 2.
Run 3.
Res:3

6. Rx Observable 對象和數據流網格

同一個項目中

  • 一部分使用了 Rx Observable 對象
  • 一部分使用了數據流網格

現在需要它們能互相溝通。

網格轉可觀察流

public static void BlockToObservableRun()
{
    var buffer = new BufferBlock<int>();
    IObservable<int> integers = buffer.AsObservable();
    integers.Subscribe(
        data => Console.WriteLine(data),
        ex => Console.WriteLine(ex),
        () => Console.WriteLine("Done"));

    buffer.Post(1);
    buffer.Post(2);
    buffer.Complete();

    buffer.Completion.Wait();
}

輸出:

1
2

AsObservable 方法會把數據流塊的完成信息（或出錯信息）轉化為可觀察流的完成信息。

  • 如果數據流塊出錯並拋出異常，這個異常信息在傳遞給可觀察流時，會被封裝在 AggregateException 對象中。

可觀察流轉網格

public static void ObservableToBlockRun()
{
    IObservable<DateTimeOffset> ticks = Observable.Interval(TimeSpan.FromSeconds(1))
        .Timestamp()
        .Select(x => x.Timestamp)
        .Take(5);

    var display = new ActionBlock<DateTimeOffset>(x => Console.WriteLine(x));
    ticks.Subscribe(display.AsObserver());

    try
    {
        display.Completion.Wait();
        Console.WriteLine("Done.");
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex);
    }
}

輸出:

2020/2/1 上午1:42:24 +00:00
2020/2/1 上午1:42:25 +00:00
2020/2/1 上午1:42:26 +00:00
2020/2/1 上午1:42:27 +00:00
2020/2/1 上午1:42:28 +00:00
Done.
  • 跟前面一樣，可觀察流的完成信息會轉化為塊的完成信息
  • 可觀察流的錯誤信息會轉化為 塊的錯誤信息。

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

-Advertisement-
Play Games
更多相關文章
  • Redis詳解（一）——RDB
    Redis詳解（一）——RDB 前言 由於 Redis 是一個記憶體資料庫，所謂記憶體資料庫，就是將資料庫中的內容保存在記憶體中，這與傳統的MySQL，Oracle等關係型資料庫直接將內容保存到硬碟中相比，記憶體資料庫的讀寫效率比傳統資料庫要快的多（記憶體的讀寫效率遠遠大於硬碟的讀寫效率）。但是保存在記憶體中也 ...
  • 親測可用！線上購書系統項目分享(Java)
    項目簡介 項目來源於： "https://gitee.com/suimz_admin/BookShop" 一個基於JSP+Servlet+Jdbc的書店系統。涉及技術少，易於理解，適合JavaWeb初學者學習使用。 本人親測可正常啟動。 技術棧 前端技術 基礎：html+css+JavaScript ...
  • SpringMVC 轉發、重定向
    轉發、重定向到其它業務方法 @org.springframework.stereotype.Controller @RequestMapping("/userController") public class UserController{ @RequestMapping("/handler1") ...
  • 任務調度
    " 返回《C 併發編程》" "1. 調度到線程池" "2. 任務調度器" "2.1. Default 調度器" "2.2. 捕獲當前同步上下文 調度器" "2.3. ConcurrentExclusiveSchedulerPair 調度器" "3. 調度並行代碼" "4. 用調度器實現數據流的同步" ...
  • .net core 常見設計模式-IChangeToken
    場景 一個對象A，希望它的某些狀態在發生改變時通知到B（或C、D），常見的做法是在A中定義一個事件（或直接用委托），當狀態改變時A去觸發這個事件。而B直接訂閱這個事件 這種設計有點問題B由於要訂閱A的事件，所以B得完全引用A，其實有時候沒必要，因為我只關心A的狀態變化而已狀態變更通知這種場景很多，有 ...
  • 取消任務
    " 返回《C 併發編程》" "1. 取消請求" "2. 超時後取消" "3. 取消並行" "4. 取消響應式代碼" "5. 與其他取消體系的互操作" 是一個等同於 預設 的特殊值，表示這個方法是永遠不會被取消的。 實例代碼 輸出: 1. 取消請求 2. 超時後取消 輸出: 只要執行代碼時用到了超時， ...
  • 線程安全集合
    " 返回《C 併發編程》" "1. 簡介" "2. 不可變棧和隊列" "3. 不可變列表" "4. 不可變Set集合" "5. 不可變字典" "6. 線程安全字典" "7. 阻塞隊列" "8. 阻塞棧和包" "9. 非同步隊列" "10. 非同步棧和包" "11. 阻塞/非同步隊列" 1. 簡介 + 不可 ...
  • c# 匿名方法（函數） 匿名委托 內置泛型委托 lamada
    匿名方法：通過匿名委托 、lamada表達式定義的函數具體操作並複製給委托類型；匿名委托：委托的一種簡單化聲明方式通過delegate關鍵字聲明；內置泛型委托：系統已經內置的委托類型主要是不帶返回值的Action和帶返回值的Func實例代碼（運行環境netcoreapp3.1)class demoF... ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類