作者：少強

在構建社交IM和朋友圈應用時，一個基本的需求是將用戶發送的消息和朋友圈更新及時準確的更新給該用戶的好友。為了做到這一點，通常需要為用戶發送的每一條消息或者朋友圈更新設置一個序號或者ID，並且保證遞增，通過這一機制來確保所有的消息能夠按照完整並且以正確的順序被接收端處理。當消息總量或者消息發送的並發數很大的時候，我們通常選擇NoSQL存儲產品來存儲消息，但常見的NoSQL產品都沒有提供自增列的功能，因此通常要藉助外部組件來實現消息序號和ID的遞增，使得整體的架構更加複雜，也影響了整條鏈路的延時。

功能介紹

表格存儲新推出的 主鍵列遞增 功能可以有效地處理上述場景的需求。具體做法為在創建表時，聲明主鍵中的某一列為自增列，在寫入一行新數據的時候，應用無需為自增列填入真實值，只需填入一個佔位符，表格存儲系統在接收到這一行數據後會自動為自增列生成一個值，並且保證在相同的分區鍵範圍內，後生成的值比先生成的值大.

主鍵列自增功能具有以下幾個特性：

• 表格存儲獨有的系統架構和主鍵自增列實現方式，可以保證生成的自增列的值唯一，且 嚴格遞增 。
• 目前支持多個主鍵，第一個主鍵為分區鍵，為了數據的均勻分布，不允許設置分區健為自增列。
• 因為分區健不允許設置為自增列，所以主鍵列自增是 分區鍵級別的自增 。
• 除了分區鍵外，其餘主鍵中的任意一個都可以被設置為遞增列。
• 對於每張表，目前 只允許設置一個主鍵列為自增列 。
• 屬性列不允許設置為自增列。
• 自增列自動生成的值為 64位的有符號長整型 。
• 自增列功能是 表級別 的，同一個實例下面可以有自增列的表，也可以有非自增列的表。
• 僅支持在創建表的時候設置自增列，對於已存在的表不支持升級為自增列。

介紹了表格存儲的主鍵列自增功能後，下面通過具體的場景介紹下如何使用。

場景

我們繼續文章開頭的例子，通過構建一個IM聊天工具，演示主鍵列自增功能的作用和使用方法。

功能

我們要做的IM聊天軟體需要支持下列功能：

• 支持用戶一對一聊天
• 支持用戶群組內聊天
• 支持同一個用戶的多終端消息同步

現有架構

第一步，確定消息模型

• 上圖展示這一消息模型
• 發送方發送了一條消息後，消息會被客戶端推送給後台系統
• 後台系統會先存儲消息
• 存儲成功後，會推送消息給接收方的客戶端

第二步，確定後台架構

• 後台架構主要分為兩部分：邏輯層和存儲層。
• 邏輯層包括應用伺服器，隊列服務和自增ID生成器，是整個後台架構的核心，處理消息的接收、推送、通知，群消息寫複製等核心業務邏輯。
• 存儲層主要是用來持久化消息數據和其他一些需要持久化的數據。
• 對於一對一聊天，發送方發送消息給應用伺服器後，應用伺服器將消息存到接收方為主鍵的表中，同時通知應用伺服器中的消息推送服務有新消息了，消息推送服務會將上次推送給接收方的最後一條消息的消息ID作為起始主鍵，從存儲系統中讀取之後的所有消息，然後將消息推送給接收方。
• 對於群組內的聊天，邏輯會更加複雜，需要通過非同步隊列來完成消息的擴散寫，也就是說發到群組內的一條消息會給群組內的每個人都存一份。

• 上圖展示了省略掉存儲層後的群消息發送過程。

• 使用擴散寫而非擴散讀，主要是由於以下兩點原因：

  • 群組內成員一般都不多，存儲成本並不高，而且有壓縮，成本更低。
  • 消息擴散寫到每個人的存儲中（收件箱）後，為每個接收方推送消息時，只需要檢查自己的收件箱即可，這時候，群聊和單聊的處理邏輯一樣，實現簡單。

• 發送方發送了一條消息後，這條消息被客戶端推送給應用伺服器，應用伺服器根據接收者的ID，將消息分發給其中一個隊列，同一個接收者的消息位於同一個隊列中，在隊列中，順序的處理每條消息，先從自增ID生成器中獲取一個新的消息ID，然後將這條消息寫入表格存儲系統。寫成功後再寫入下一條消息。

• 同一個接收方的消息會盡量在一個隊列中，一個隊列中可能會有多個接收方的消息。

• 群組內聊天時可能會出現同一個時刻兩個用戶同時發送了消息，這兩個消息可能會進入不同的應用伺服器，但是應用伺服器會將同一個接收方的消息發給同一個隊列服務，這時候，對於同一個接收方，這兩條消息就會處於同一個隊列中，如下圖：

• 每個隊列中的數據串列處理，每次寫入表格存儲的時候，分配一個新的ID，比之前的ID要大，為了保證消息可以嚴格遞增，避免前一個消息寫失敗導致無法嚴格遞增的情況出現，需要在寫入數據到存儲系統的時候，持有一個用戶級別的鎖，在沒有寫成功之前，同用戶的其他消息不能繼續寫，以免當前消息寫失敗後導致亂序，當寫成功後，釋放這個鎖，下一個消息繼續。

• 上一步中，如果隊列宕機，這些消息需要重新處理，這時候，原有消息就會進入一個新的隊列，這時候新的隊列需要一個新的消息ID，但要比之前已有的消息ID更大，而這個新隊列並不知道之前的最大ID是啥，所以，這裡每個隊列沒法自主創建自增ID，而需要一個全局的自增ID生成器。

• 為了支持多終端，在應用伺服器中會為每個終端持有一個session，每個session持有一個當前最新消息的ID，當被通知有新消息時，會去存儲系統讀取當前消息之後的所有消息，這樣就保證了多終端同時在線時，每個終端都可以同步消息，且相互不影響，見下圖。

• 在多終端中，如果有部分終端由在線變成了離線，那麼應用伺服器會將這個終端的session保存到存儲系統的另一張表中，當一段時間後，這個終端再次上線時，可以從存儲系統中恢復出之前的session，繼續為此終端推送之前未讀取的消息。

第三步，確定存儲系統

存儲系統，我們選擇了阿里雲的 表格存儲 ，主要是因為下列原因：

• 寫操作不僅支持 單行寫 ，也支持 多行批量寫 ，滿足大並發寫數據需求。
• 支持按 範圍讀 ，消息多時可翻頁。
• 支持 數據生命周期管理 ，對過期數據進行自動清理，節省存儲費用，詳細文檔
• 表格存儲是阿里雲已經商業化的雲服務， 穩定可靠 。
• 表格存儲 價格便宜，對於數據量大的用戶還可以以更優惠的價格購買套餐。
• 讀寫性能優秀，對於聊天消息，延遲基本在毫秒，甚至微妙級別。

第四步，確定表結構

確定的表格存儲的表結構如下：

• 到此，我們已經設計出了一個完整的聊天系統，雖然這個系統已經可以運行，且能處理大並發，性能也不差，但是還是存在一些挑戰。表格存儲的表結構分為兩部分，主鍵列部分和屬性列部分，主鍵列部分最多支持4個主鍵，第一個主鍵為分區健。
• 使用前，需要確定主鍵列部分的結構，使用過程中不能修改；屬性列部分是Schema Free的，用戶可以自由定製，每一行數據的屬性列部分可以不一樣，所以，只需要設計主鍵列部分的結構。
• 第一個主鍵是分片鍵，目的是讓數據和請求可以均衡分布，避免熱點，由於最終讀取消息時是要按照接收方讀取，所以這裡可以使用接收方ID作為分片鍵，為了更加均衡，可以使用接收方ID的md5值的部分區域，比如前4個字元。這樣就可以將數據均衡分布了。
• 第一個主鍵只用了部分接收方ID，為了能定位到接收方的消息，需要保存完整的接收方ID，所以，可以將接收方ID作為第二個主鍵。
• 第三個主鍵就可以是消息ID了，由於需要查詢最新的消息，這個值需要是單調自增的。
• 屬性列可以存消息內容和元數據等。

挑戰

• 多個用戶在一個隊列中，這個隊列串列執行，為了保證消息嚴格遞增，這裡執行過程中要持有鎖，這裡就會有一個風險點：如果發送給某個用戶的消息量很大，這個用戶所在的隊列中消息會變多，就有可能堵塞其他用戶的消息，導致同隊列的其他用戶消息出現延遲。
• 當出現重大事件或者特定節假日，聊天信息量大的時候，隊列部分需要擴容，否則可能扛不住大壓力，導致整體系統延遲增大或者崩潰。

針對上述兩個問題，問題2可以通過增加機器的方式解決，但是問題1沒法通過增加機器解決，增加機器只能緩解問題，卻沒法徹底解決。那有沒有辦法可以徹底解決掉上述兩個問題？

新架構

上面兩個問題的複雜度主要是由於需要消息嚴格遞增引起的，如果使用了表格存儲的主鍵列自增功能，那麼上層的應用層就會簡單的多。

使用了表格存儲**主鍵列自增功能**後的新架構如下：

• 最明顯的區別是少了隊列服務和自增ID生成器兩個組件，架構更加簡單。
• 應用伺服器接收到消息後，直接將消息寫入表格存儲，對於主鍵自增列message_id，在寫數據時不需要填確定的值，只需要填充一個特定的佔位符即可，這個值會在表格存儲系統內部自動生成。
• 新架構中自增操作是在表格存儲系統內部處理的，就算多個應用伺服器同時給表格存儲中的同一個接收方寫數據，表格存儲內部也能保證這些消息是串列處理，每個消息都有一個獨立的消息ID，且嚴格遞增。那麼之前的隊列服務就不在需要了。這樣也就 徹底解決了上面的問題1
• 表格存儲系統是一個雲服務，用戶並不需要考慮系統的容量，而且表格存儲支持按量付費，這樣也就 徹底解決了上面的問題2
• 之前只能有一個隊列處理同一個用戶的消息，現在可以多個隊列並行處理了，就算某些用戶的消息量突然變大，也不會立即堵塞其他用戶，而是將壓力均勻分布給了所有隊列。
• 使用主鍵自增列功能後，應用伺服器可以直接寫數據到表格存儲，不再需要經過隊列和獲取消息ID， 性能表現會更加優秀。

實現

有了上面的架構圖後，現在可以開始實現了，這裡選用JAVA SDK，目前4.2.0版本已經支持主鍵列自增功能，4.2.0版本Java SDK文檔和下載地址。

第一步，建表

按照之前的設計，表結構如下：

第三列PK是message_id，這一列是主鍵自增列，建表時指定message_id列的屬性為AUTO_INCREMENT，且類型為INTEGER。

private static void createTable(SyncClient client) {n TableMeta tableMeta = new TableMeta(「message_table」);nn // 第一列為分區建n tableMeta.addPrimaryKeyColumn(new PrimaryKeySchema("partition_key", PrimaryKeyType.STRING));nn // 第二列為接收方IDn tableMeta.addPrimaryKeyColumn(new PrimaryKeySchema("receive_id", PrimaryKeyType.STRING));nn // 第三列為消息ID，自動自增列，類型為INTEGER，屬性為PKO_AUTO_INCREMENTn tableMeta.addPrimaryKeyColumn(new PrimaryKeySchema("message_id", PrimaryKeyType.INTEGER, PrimaryKeyOption.AUTO_INCREMENT));nn int timeToLive = -1; // 永不過期，也可以設置數據有效期，過期了會自動刪除n int maxVersions = 1; // 只保存一個版本，目前支持多版本nn TableOptions tableOptions = new TableOptions(timeToLive, maxVersions);nn CreateTableRequest request = new CreateTableRequest(tableMeta, tableOptions);nn client.createTable(request);n }n

通過上述方式就創建了一個第三列PK為自動自增的表。

第二步，寫數據

寫數據目前支持PutRow和BatchWriteRow兩種方式，這兩種介面都支持主鍵列自增功能，寫數據時，第三列message_id是主鍵自增列，這一列不需要填值，只需要填入佔位符即可。

private static void putRow(SyncClient client, String receive_id) {n // 構造主鍵n PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();nn // 第一列的值為 hash(receive_id)前4位n primaryKeyBuilder.addPrimaryKeyColumn(「partition_key」, PrimaryKeyValue.fromString(hash(receive_id).substring(4)));nn // 第二列的值為接收方IDn primaryKeyBuilder.addPrimaryKeyColumn(「receive_id」, PrimaryKeyValue.fromString(receive_id));nn // 第三列是消息ID，主鍵遞增列，這個值是TableStore產生的，用戶在這裡不需要填入真實值，只需要一個佔位符：AUTO_INCREMENT 即可。n primaryKeyBuilder.addPrimaryKeyColumn("message_id", PrimaryKeyValue.AUTO_INCREMENT);n PrimaryKey primaryKey = primaryKeyBuilder.build();nn RowPutChange rowPutChange = new RowPutChange("message_table", primaryKey);nn // 這裡設置返回類型為RT_PK，意思是在返回結果中包含PK列的值。如果不設置ReturnType，默認不返回。n rowPutChange.setReturnType(ReturnType.RT_PK);nn //加入屬性列，消息內容n rowPutChange.addColumn(new Column("content", ColumnValue.fromString(content)));nn //寫數據到TableStoren PutRowResponse response = client.putRow(new PutRowRequest(rowPutChange));nn // 列印出返回的PK列n Row returnRow = response.getRow();n if (returnRow != null) {n System.out.println("PrimaryKey:" + returnRow.getPrimaryKey().toString());n }nn // 列印出消耗的CUn CapacityUnit cu = response.getConsumedCapacity().getCapacityUnit();n System.out.println("Read CapacityUnit:" + cu.getReadCapacityUnit());n System.out.println("Write CapacityUnit:" + cu.getWriteCapacityUnit());n }n

第三步，讀數據

讀消息的時候，需要通過GetRange介面讀取最近的消息，message_id這一列PK的起始位置是上一條消息的message_id+1, 結束位置是INF_MAX，這樣每次都可以讀出最新的消息，然後發送給客戶端

private static void getRange(SyncClient client, String receive_id, String lastMessageId) {n RangeRowQueryCriteria rangeRowQueryCriteria = new RangeRowQueryCriteria(「message_table」);nn // 設置起始主鍵n PrimaryKeyBuilder primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();nn // 第一列的值為 hash(receive_id)前4位n primaryKeyBuilder.addPrimaryKeyColumn(「partition_key」, PrimaryKeyValue.fromString(hash(receive_id).substring(4)));nn // 第二列的值為接收方IDn primaryKeyBuilder.addPrimaryKeyColumn(「receive_id」, PrimaryKeyValue.fromString(receive_id));nn // 第三列的值為消息ID，起始於上一條消息n primaryKeyBuilder.addPrimaryKeyColumn(「message_id」, PrimaryKeyValue.fromLong(lastMessageId + 1));n rangeRowQueryCriteria.setInclusiveStartPrimaryKey(primaryKeyBuilder.build());nn // 設置結束主鍵n primaryKeyBuilder = PrimaryKeyBuilder.createPrimaryKeyBuilder();nn // 第一列的值為 hash(receive_id)前4位n primaryKeyBuilder.addPrimaryKeyColumn(「partition_key」, PrimaryKeyValue.fromString(hash(receive_id).substring(4)));nn // 第二列的值為接收方IDn primaryKeyBuilder.addPrimaryKeyColumn(「receive_id」, PrimaryKeyValue.fromString(receive_id));nn // 第三列的值為消息IDn primaryKeyBuilder.addPrimaryKeyColumn("message_id", PrimaryKeyValue.INF_MAX);n rangeRowQueryCriteria.setExclusiveEndPrimaryKey(primaryKeyBuilder.build());nn rangeRowQueryCriteria.setMaxVersions(1);nn System.out.println("GetRange的結果為:");n while (true) {n GetRangeResponse getRangeResponse = client.getRange(new GetRangeRequest(rangeRowQueryCriteria));n for (Row row : getRangeResponse.getRows()) {n System.out.println(row);n }nn // 若nextStartPrimaryKey不為null, 則繼續讀取.n if (getRangeResponse.getNextStartPrimaryKey() != null) {n rangeRowQueryCriteria.setInclusiveStartPrimaryKey(getRangeResponse.getNextStartPrimaryKey());n } else {n break;n }n }n }n

上面演示了表格存儲及其主鍵列自增功能在聊天系統中的應用，在其他場景中也有很大的價值，期待大家一起去探索。