反應式程式設計在客戶端程式設計當中的應用相當廣泛，而當前在服務端中的應用相對被提及較少。本篇將介紹如何在服務端程式設計中應用響應時程式設計來改進資料庫操作的效能。

開篇就是結論

接續上一篇《談反應式程式設計在服務端中的應用，資料庫操作優化，從 20 秒到 0.5 秒》之後，這次，我們帶來了關於利用反應式程式設計進行 upsert 優化的案例說明。建議讀者可以先閱讀一下前一篇，這樣更容易理解本篇介紹的方法。

同樣還是利用批量化的思路，將單個 upsert 操作批量進行合併。已達到減少資料庫連結消耗從而大幅提升效能的目的。

業務場景

在最近的一篇文章《十萬同時線上使用者，需要多少記憶體？——Newbe.Claptrap 框架水平擴充套件實驗》中。我們通過啟用多個常駐於記憶體當中的 Claptrap 來實現快速驗證 JWT 正確性的目的。

但，當時有一個技術問題沒有得到解決：

Newbe.Claptrap 框架設計了一個特性：當 Claptrap Deactive 時，可以選擇將快照立即儲存到資料庫。因此，當嘗試從叢集中關閉一個節點時，如果節點上存在大量的 Claptrap ，那麼將產生大量的資料庫 upsert 操作。瞬間推高資料庫消耗，甚至導致部分錯誤而儲存失敗。

一點點程式碼

有了前篇的 IBatchOperator，那麼留給這篇的程式碼內容就非常少了。

首先，按照使用上一篇的 IBatchOperator 編寫一個支援操作的 Repository，形如以下程式碼：

public class BatchUpsert : IUpsertRepository
{
    private readonly IDatabase _database;
    private readonly IBatchOperator<(int, int), int> _batchOperator;

    public BatchUpsert(IDatabase database)
    {
        _database = database;
        var options = new BatchOperatorOptions<(int, int), int>
        {
            BufferCount = 100,
            BufferTime = TimeSpan.FromMilliseconds(50),
            DoManyFunc = DoManyFunc
        };
        _batchOperator = new BatchOperator<(int, int), int>(options);
    }

    private Task<int> DoManyFunc(IEnumerable<(int, int)> arg)
    {
        return _database.UpsertMany(arg.ToDictionary(x => x.Item1, x => x.Item2));
    }

    public Task UpsertAsync(int key, int value)
    {
        return _batchOperator.CreateTask((key, value));
    }
}

然後，只要實現對應資料庫的 UpsertMany 方法，便可以很好地完成這項優化。

各種資料庫的操作

結合 Newbe.Claptrap 現在專案的實際。目前，被支援的資料庫分別有 SQLite、PostgreSQL、MySql 和 MongoDB。以下，分別對不同型別的資料庫的批量 Upsert 操作進行說明。

由於在 Newbe.Claptrap 專案中的 Upsert 需求都是以主鍵作為對比鍵，因此以下也只討論這種情況。

SQLite

根據官方文件，使用 INSERT OR REPLACE INTO 便可以實現主鍵衝突時替換資料的需求。

具體的語句格式形如以下：

INSERT OR REPLACE INTO TestTable (id, value)
VALUES
(@id0,@value0),
...
(@idn,@valuen);

因此只要直接拼接語句和引數呼叫即可。需要注意的是，SQLite 的可傳入引數預設為 999，因此拼接的變數也不應大於該數量。

官方文件：INSERT

PostgreSQL

眾所周知，PostgreSQL 在進行批量寫入時，可以使用高效的 COPY 語句來完成資料的高速匯入，這遠遠快於 INSERT 語句。但可惜的是 COPY 並不能支援 ON CONFLICT DO UPDATE 子句。因此，無法使用 COPY 來完成 upsert 需求。

因此，我們還是迴歸使用 INSERT 配合 ON CONFLICT DO UPDATE 子句，以及 unnest 函式來完成批量 upsert 的需求。

具體的語句格式形如以下：

INSERT INTO TestTable (id, value)
VALUES (unnest(@ids), unnest(@values))
ON CONFLICT ON CONSTRAINT TestTable_pkey
DO UPDATE SET value=excluded.value;

其中的 ids 和 values 分別為兩個等長的陣列物件，unnest 函式可以將陣列物件轉換為行資料的形式。

注意，可能會出現 ON CONFLICT DO UPDATE command cannot affect row a second time 錯誤。

因此如果嘗試使用上述方案，需要在傳入資料庫之前，先在程式中去重一遍。而且，通常來說，在程式中進行一次去重可以減少向資料庫中傳入的資料，這本身也很有意義。

官方文件：unnest 函式
官方文件：Insert 語句

MySql

MySql 與 SQLite 類似，支援 REPLACE 語法。具體語句形式如下：

REPLACE INTO TestTable (id, value)
VALUES
(@id0,@value0),
...
(@idn,@valuen);

官方文件：REPLACE 語句

MongoDB

MongoDB 原生支援 bulkWrite 的批量傳輸模式，也支援 replace 的 upsert 語法。因此操作非常簡單。

那麼這裡展示一下 C# 操作方法：

private async Task SaveManyCoreMany(
    IDbFactory dbFactory,
    IEnumerable<StateEntity> entities)
{
    var array = entities as StateEntity[] ?? entities.ToArray();
    var items = array
        .Select(x => new MongoStateEntity
        {
            claptrap_id = x.ClaptrapId,
            claptrap_type_code = x.ClaptrapTypeCode,
            version = x.Version,
            state_data = x.StateData,
            updated_time = x.UpdatedTime,
        })
        .ToArray();

    var client = dbFactory.GetConnection(_connectionName);
    var db = client.GetDatabase(_databaseName);
    var collection = db.GetCollection<MongoStateEntity>(_stateCollectionName);

    var upsertModels = items.Select(x =>
    {
        var filter = new ExpressionFilterDefinition<MongoStateEntity>(entity =>
            entity.claptrap_id == x.claptrap_id && entity.claptrap_type_code == x.claptrap_type_code);
        return new ReplaceOneModel<MongoStateEntity>(filter, x)
        {
            IsUpsert = true
        };
    });
    await collection.BulkWriteAsync(upsertModels);
}

這是從 Newbe.Claptrap 專案業務場景中給出的程式碼，讀者可以結合自身需求進行修改。

官方文件：db.collection.bulkWrite ()

通用型解法

優化的本質是減少資料庫連結的使用，儘可能在一個連結內完成更多的工作。因此如果特定的資料庫不支援以上資料庫類似的操作。那麼還是存在一種通用型的解法：

1. 以儘可能快地方式將資料寫入一臨時表
2. 將臨時表的資料已連表 update 的方式更新的目標表
3. 刪除臨時表

UPDATE with a join

效能測試

以 SQLite 為例，嘗試對 12345 條資料進行 2 次 upsert 操作。

單條併發：1 分 6 秒

批量處理：2.9 秒

可以在該連結找到測試的程式碼。

樣例中不包含有 MySql、PostgreSQL 和 MongoDB 的樣例，因為沒有優化之前，在不提高連線池的情況下，一併發基本就爆炸了。所有優化的結果是直接解決了可用性的問題。

所有的示例程式碼均可以在程式碼庫中找到。如果 Github Clone 存在困難，也可以點選此處從 Gitee 進行 Clone

常見問題解答

此處對一些常見的問題進行解答。

客戶端是等待批量操作的結果嗎？

這是一個很多網友提出的問題。答案是：是的。

假設我們公開了一個 WebApi 作為介面，由瀏覽器呼叫。如果同時有 100 個瀏覽器同時發出請求。

那麼這 100 個請求會被合併，然後寫入資料庫。而在寫入資料庫之前，這些客戶端都不會得到服務端的響應，會一直等待。

這也是該合併方案區別於普通的 “寫佇列，後寫庫” 方案的地方。

原理上講，這種和 bulkcopy 有啥不一樣？

兩者是不相關，必須同時才有作用的功能。
首先，程式碼中的 database.InsertMany 就是你提到的 bulkcopy。

這個程式碼的關鍵不是 InsertMany ，而是如何將單次的插入請求合併。
試想一下，你可以在 webapi 上公開一個 bulkcopy 的 API。
但是，你無法將來自不同客戶端的請求合併在同一個 API 裡面來呼叫 bulkcopy。
例如，有一萬個客戶端都在呼叫你的 API，那怎麼合併這些 API 請求呢？

如果如果通過上面這種方式，雖然你只是對外公開了一個單次插入的 API。你卻實現了來自不同客戶端請求的合併，變得可以使用 bulkcopy 了。這在高併發下很有意義。

另外，這符合開閉的原理，因為你沒有修改 Repository 的 InsertOne 介面，卻實現了 bulkcopy
的效果。

如果批量操作中一個操作異常失敗是否會導致被合併的其他操作全部失敗？

如果業務場景是合併會有影響，那當然不應該合併。

批量操作一個失敗，當然是一起失敗，因為底層的資料庫事務肯定也是一起失敗。

除非批量介面也支援對每個傳入的 ID 做區別對待。典型的，比如 mongodb 的 bulkcopy 可以返回哪些成功哪些失敗，那麼我們就有能力設定不同的 Tcs 狀態。

哪些該合併，哪些不該合併，完全取決於業務。樣例給出的是如果要合併，應該怎麼合併。不會要求所有都要合併。

Insert 和 Upsert 都說了，那 Delete 和 Select 呢？

筆者籠統地將該模式稱為 “反應式批量處理”。要確認業務場景是否應用該模式，需要具備以下這兩個基本的要求：

• 業務下游的批量處理是否會比累積的單條處理要快，如果會，那可以用
• 業務上游是否會出現短時間的突增頻率的請求，如果會，那可以用

當然，還需要考量，比如：下游的批量操作能否卻分每個請求的結果等等問題。但以上兩點是一定需要考量的。

那麼以 Delete 為例：

• Delete Where In 的速度會比 Delete = 的速度快嗎？試一下
• 會有突增的 Delete 需求嗎？想一下

小小工具 Zeal

筆者是一個完整儲存過程都寫不出來的人。能夠查閱到這些資料庫的文件，全靠一款名為 Zeal 的離線文件檢視免費軟體。推薦給您，您也值得擁有。

Zeal 官網地址：https://zealdocs.org/

最後但是最重要！

最近作者正在構建以反應式、Actor模式和事件溯源為理論基礎的一套服務端開發框架。希望為開發者提供能夠便於開發出 “分散式”、“可水平擴充套件”、“可測試性高” 的應用系統 ——Newbe.Claptrap

本篇文章是該框架的一篇技術選文，屬於技術構成的一部分。如果讀者對該內容感興趣，歡迎轉發、評論、收藏文章以及專案。您的支援是促進專案成功的關鍵。

如果你對該專案感興趣，你可以通過 github issues 提交您的看法。

如果您無法正常訪問 github issue，您也可以傳送郵件到 [email protected] 來參與我們的討論。

點選連結 QQ 交流【Newbe.Claptrap】：https://jq.qq.com/?_wv=1027&k=5uJGXf5。

您還可以查閱本系列的其他選文：

• Newbe.Claptrap - 一套以 “事件溯源” 和 “Actor 模式” 作為基本理論的服務端開發框架
• 十萬同時線上使用者，需要多少記憶體？——Newbe.Claptrap 框架水平擴充套件實驗
• 談反應式程式設計在服務端中的應用，資料庫操作優化，從 20 秒到 0.5 秒
• 談反應式程式設計在服務端中的應用，資料庫操作優化，提速 Upsert
• Newbe.Claptrap 專案週報 1 - 還沒輪影，先用輪跑

GitHub 專案地址：https://github.com/newbe36524/Newbe.Claptrap

Gitee 專案地址：https://gitee.com/yks/Newbe.Claptrap

• 本文作者： newbe36524
• 本文連結： https://www.newbe.pro/Newbe.Claptrap/Reactive-In-Server-2/
• 版權宣告： 本部落格所有文章除特別宣告外，均採用 BY-NC-SA 許可協議。轉載請註明出處！
  相關推薦

  談反應式程式設計在服務端中的應用，資料庫操作優化，萬條記錄從20秒到0.5秒

  反應式程式設計在客戶端程式設計當中的應用相當廣泛，而當前在服務端中的應用相對被提及較少。本篇將介紹如何在服務端程式設計中應用響應時程式設計來改進資料庫操作的效能。 開篇就是結論 利用 System.Reactive 配合 TaskCompelteSource ，可以將分散的單次資料庫插入請求合併會一個批量插入

  談反應式程式設計在服務端中的應用，資料庫操作優化，提速 Upsert

  反應式程式設計在客戶端程式設計當中的應用相當廣泛，而當前在服務端中的應用相對被提及較少。本篇將介紹如何在服務端程式設計中應用響應時程式設計來改進資料庫操作的效能。 開篇就是結論 接續上一篇《談反應式程式設計在服務端中的應用，資料庫操作優化，從 20 秒到 0.5 秒》之後，這次，我們帶來了關於利用反應式程式設

  響應式程式設計在Android中的應用

  響應式程式設計簡介 響應式程式設計是一種基於非同步資料流概念的程式設計模式。資料流就像一條河：它可以被觀測，被過濾，被操作，或者為新的消費者與另外一條流合併為一條新的流。 響應式程式設計的一個關鍵概念是事件。事件可以被等待，可以觸發過程，也可以

  QQ 幫助他人-01-有人遇到過Skype服務端中出現MCU報錯的問題麽？導致無法實現多方視頻?

  服務器 Lync 【初中】北京-XX 2017/12/21 17:37:52 求助一下：有人遇到過SKYPE服務端中出現MCU報錯的問題麽？導致無法實現多方視頻 【管理員】上海-學無止境(3313395633) 2017/12/21 17:41:13 以前正常？ 【初中】北京-XXX 2017/12

  SpringCloud分布式事務實戰（七）在微服務1中創建整合函數，調用微服務2

  request enable class alt cef 內容 llb 傳遞 turn （1） 添加jar pom.xml <dependency> <groupId>org.springframework.clou

  python 學習筆記_2 模擬socket程式設計 服務端、客戶端通訊(參考核心程式設計2程式碼實現)

  伺服器端程式碼實現： #!/usr/bin/env python#coding=gbk'''接收客戶端字串，在欄位串前面打上當前時間，然後返回server端採用 python2 linux下除錯執行客戶端採用python3 windows下除錯執行（注意編碼格式）''' from socket impor

  響應式程式設計在Android 中的一些探索

      響應式在前端領域已經變得十分流行，很多主流框架都採用響應式來進行頁面的展示重新整理。本文主要是探索一下響應式在移動端Android上的一些實踐，包括對響應式思想的理解，以及目前Android上實現響應式的一些手段，最後聊聊響應式在Android開發上的一些應用。

  也談如何構建高效能服務端程式

  引子：我接觸過很多程式語言，接觸過各種各樣的伺服器端開發，Java，Go，Ruby，Javascript等語言，Spring，Node.js，Rails等等常見伺服器端框架和程式設計模型都有接觸。這裡談一下我個人對高效能伺服器端程式的一些看法，希望給各位讀者一些認識。這片文章提到的內容也是 Coding（ht

  反應式程式設計之RxJava

  Subject介紹： Table of Contents AsyncSubject:Observer會接收AsyncSubject的onComplete()之前的最後一個數據 BehaviorSubject:Observer會接收到BehaviorSubje

  Rxjava2入門教程二：Observable與Observer響應式程式設計在Rxjava2中的典型實現

  在RxJava中，函式響應式程式設計具體表現為一個觀察者（Observer）訂閱一個可觀察物件（Observable），通過建立可觀察物件發射資料流，經過一系列操作符（Operators）加工處理和執行緒排程器（Scheduler）在不同執行緒間的轉發，最後由觀察者接受並做出響應的一個過程 Observ

  非同步服務端中的多執行緒 boost asio

  這個demo程式好像會死鎖，哪位高手要是能找到原因麻煩告訴一下 // gogogo.cpp : Defines the entry point for the console application. // #include "stdafx.h" #ifdef WIN3

  Socket程式設計服務端和客戶端互相通訊，實現簡單的聊天

  客戶端兩個執行緒分別記錄讀和寫，服務端兩個執行緒分別記錄讀和寫； 客戶端程式碼： import java.io.BufferedReader; import java.io.BufferedWriter; import java.io.IOException; import

  Java 平臺反應式程式設計（Reactive Programming）入門

  最近的一段時間裡，反應式程式設計在社群中得到了很大的關注。從Java社群來說，Java 9把反應式流規範以java.util.concurrent.Flow 類的形式新增到了標準庫中。Spring 5 已經支援了反應式程式設計實踐，並提供了 WebFlux 這樣的 Web 程

  指令式程式設計 vs 函數語言程式設計 vs 反應式程式設計

  指令式程式設計 存在潛在的阻塞。 函數語言程式設計 避免處理資料的中間狀態。可以方便的處理資料流。 反應式程式設計 多種方式使用資料流，合併，過濾和轉換等。 抽象層次更高，關注於如何把小的方法

  Redis在三層服務框架中應用（一）——Redis與Model的結合

  個人宣告:本系列所有文章旨在拋磚引玉，為有興趣深入使用Redis的同學提供一些參考。本系列所有文章純屬原創，均是筆者在實際工作中的總結。本文所有引用的MDB系列元件均由米多網路架構部提供，在此向架構部表示感謝。筆者經驗能力有限，如有不適之處還請多多指教。一、為什麼要使用Red

  windows socket程式設計(服務端頻繁的執行FD_CLOSE)

  if(NetworkEvents.lNetworkEvents & FD_CLOSE){pGenericServer->CleanupThread(Event, s, pNewConn, GetCurrentThreadId());#ifdef GENERICSERVER_DEBUGOUTch

  淺談契約式程式設計

  　　契約式程式設計是程式設計的一種方法。那麼什麼是契約式程式設計呢？我想這個概念是從“合同”演變過來的。 　　在人類的社會活動中，契約一般是用於兩方，一方（供應者）為另一方（客戶）完成一些任務。每一方都期待從契約中獲得利益，同時也要接受一些義務。通常，一方視為義務的對另一方來說是權利。契約文件要清楚地寫明雙

  淺談鏈式程式設計及建造者模式

  今天談談鏈式程式設計,和建造者模式，第一部分談鏈式程式設計，第二部分談建造者模式。 在安卓開發中，最早接觸鏈式程式設計，應該算是AlertDialog類了，後來在github上看到JakeWharton 的picasso圖片處理框架。從ViewpageIndi

  .NET併發程式設計-反應式程式設計

  本系列學習在.NET中的併發並行程式設計模式，實戰技巧 本小節開始學習反應式程式設計。本系列保證最少程式碼呈現量，雖然talk is cheap, show me the code被奉為圭臬，我的學習習慣是，只學習知識點，程式碼不在當下立馬要用的時候不會認真去讀的，更何況在大多時候在手機閱讀更不順暢。

  Mac下如何用SSH連接遠程Linux服務器及Linux一些常用操作命令，更新中.....

  oot .... 地址 密碼 res ini 更新 目錄 讀寫 1、 終端命令 　　a).打開Mac的命令終端 　　b).輸入ssh -p 22 [email protected] 它會提示你輸入密碼,輸入正確的密碼之後,你就發現已經登陸成功了.(22: 端口號 roo