最近有一個項目，其中某個功能單表數據在可預估的未來達到了億級，初步估算在90億左右。與同事詳細討論后，決定采用一致性Hash算法來完成數據庫的自動擴容和數據遷移。整個程序細節由我同事完成，我只是將其理解並成文，供有相同問題的同行參考。
參看此文的兄弟，默認各位已經熟悉一致性hash算法了。此文僅僅闡述代碼細節，實現語言為Java。
項目背景
- 項目是一個實驗室項目
- 其中有一個表叫做試驗表，用於存儲車型的試驗數據，每個試驗大概有6000條數據
- 總計初期約有2萬個車型，每個車型初期包含超過50個試驗。后期還會動態增長
- 試驗表中的數據僅需要根據車型試驗ID能取出來即可，沒有其他更復雜的業務邏輯
方案決策
項目正式上線初期，數據量不會直接爆發式增長到90億，需要時間上的積累（逐步做實驗），最終可能達到90億數據，甚至超過90億數據。
按照我們實際了解情況，oracle存儲數據量達到1千萬的時候，性能擅可。而Oracle官方的說法，如單表存儲1g有分區（大致500萬數據），查詢效率非常高。而試驗表中僅四個字段，每條數據數據量較小。所以我們最終決定以1000萬為節點，水平拆表。當表數據達到1千萬時，即增加下一波表。進行數據自動遷移。
按照90億的總量，1000萬數據一個表的划分，最終大致會產生900個左右的表。所以我們最終使用了4個數據庫。1個存儲其他業務模塊的表，3個存儲此大數據表。每個數據庫大致有300張表。性能上和數量上都可達到我們的要求。
相關表結構
試驗信息表（EXPERIMENT_MESSAGE），掛接車型和試驗的關系。試驗數據表（EXPERIMENT_DATA），存儲試驗數據
試驗信息表：
| 字段 | 含義 |
|---|---|
| ID | 主鍵，采用UUID生成 |
| EXPERIMENT_ID | 試驗表中的ID |
| CAR_ID | 車型表中的ID |
| ... | 其余數十個字段省略 |
試驗數據表：
| 字段 | 含義 |
|---|---|
| ID | 主鍵，采用UUID生成 |
| EXPERIMENT_MESSAGE_ID | 對應的實驗信息id |
| X_VALUE | 試驗數據X值 |
| Y_VALUE | 試驗數據Y值 |
我們采用作一致性hash的key，就是試驗數據表中的EXPERIMENT_MESSAGE_ID字段。也就是說，每個試驗數據表，不存則以，存則一次性大致有6000條數據。取同理。
一致性Hash算法實現
一致性Hash算法的hash部分，采用了著名的ketama算法。在此，我們不多討論ketama算法的細節，若各位有興趣，請查閱ketama算法
public long hash(String key) {
if (md5 == null) {
try {
md5 = MessageDigest.getInstance("MD5");
} catch (NoSuchAlgorithmException e) {
throw new IllegalStateException("no md5 algorythm found");
}
}
md5.reset();
md5.update(key.getBytes());
byte[] bKey = md5.digest();
long res = ((long) (bKey[3] & 0xFF) << 24) |
((long) (bKey[2] & 0xFF) << 16) |
((long) (bKey[1] & 0xFF) << 8) |
(long) (bKey[0] & 0xFF);
return res & 0xffffffffL;
}
有了Hash的算法，接下來就要構造Hash環了。Hash環采用的SortedMap數據結構實現。
private final SortedMap<Long, T> circle = new TreeMap<Long, T>();
其中添加節點和移除節點部分，需要根據hash算法得到節點在環上的位置，具體代碼如下：
/**
* 添加虛擬節點
* numberOfReplicas為虛擬節點的數量，初始化hash環的時候傳入，我們使用300個虛擬節點
* @param node
*/
public void add(T node) {
for (int i = 0; i < numberOfReplicas; i++) {
circle.put(hashFunction.hash(node.toString() + i), node);
}
}
/**
* 移除節點
* @param node
*/
public void remove(T node) {
for (int i = 0; i < numberOfReplicas; i++) {
circle.remove(hashFunction.hash(node.toString() + i));
}
}
而hash環中得到節點部分比較特殊，根據一致性hash算法的介紹，得到hash環中的節點，實際上是計算出的hash值順時針找到的第一個節點。
/**
* 獲得一個最近的順時針節點
* @param key 為給定鍵取Hash，取得順時針方向上最近的一個虛擬節點對應的實際節點
* @return
*/
public T get(Object key) {
if (circle.isEmpty()) {
return null;
}
long hash = hashFunction.hash((String) key);
if (!circle.containsKey(hash)) {
//返回此映射的部分視圖，其鍵大於等於 hash
SortedMap<Long, T> tailMap = circle.tailMap(hash);
hash = tailMap.isEmpty() ? circle.firstKey() : tailMap.firstKey();
}
return circle.get(hash);
}
單表拆分實踐
上面完成了一致性hash算法的實現，包含了hash算法和hash環的實現。接下來就要處理具體業務中，如何使用這個hash環和算法了。
我們業務中，主要操作這張表的數據，也就是增刪查。然后我們數據庫拆分成了3個，所以需要增刪查的操作基本一致，都是先通過一致性hash得到庫，再通過一致性hash得到表。
獲取數據庫名的操作如下，獲取到數據庫后，根據數據庫名到對應的連接池中獲取連接。
/**
* 根據試驗信息id獲取其所在庫名
* DatabaseType為我們數據的枚舉
* @return 數據庫的名稱
**/
private String getDataBase(String experimentMessageId) {
//獲取數據源
DatabaseType[] databasetype = DatabaseType.values();
List<String> dataBaselist = new ArrayList<>();
Map<String, DatabaseType> map = new HashMap<>();
for (DatabaseType d:databasetype) {
if (!d.equals(DatabaseType.KC)) {
dataBaselist.add(d.toString());
map.put(d.toString(), d);
}
}
//獲取數據源hash
ConsistentHash<String> dataBaseCon = getConsistentHash(dataBaselist);
//獲取id所在數據源
String dataBase = dataBaseCon.get(experimentMessageId);
return dataBase;
}
獲取表名的操作如下，獲取到數據庫后，在對應的數據庫中找到需要的表，再從該表中查詢數據。
/**
* 根據試驗信息id獲取其試驗數據所在表
* @return
**/
public String getTableName(String experimentMessageId) {
String dataBase = getDataBase(experimentMessageId);
//查詢所有試驗數據表
List<String> tables = experimentDataEODao.queryTbaleNames(dataBase, tableName);
ConsistentHash<String> consistentHash = getConsistentHash(tables);
String tableName = consistentHash.get(experimentMessageId);
return tableName;
}
剩下的增刪改操作和平常一致，在此不多贅述。
數據遷移實踐
一致性hash勢必涉及到數據遷移問題，我們采取的數據遷移方式為定時任務，針對每個數據庫在每天夜里全量掃描一次。檢查是否有數據量超過1000萬的表，若存在這樣的表，就把現有的表數量double。
數據遷移只會在同庫之間遷移，不會涉及跨數據庫的情況。
此方案為初步方案，后續會改進的更加智能，根據表的數量，增加不同數量的表。而不是簡單的把表數量翻倍。
表創建后，將需要遷移的表數據逐個遷移。
在連接到數據源后，我們做了如下事情進行數據遷移
1.獲取庫中所有的表
List<String> tables = getTables(connection, p, d.toString());
2.遍歷表，檢查表中數據是否超過邊界線（我們為1000萬）
for (int i = 0; i < tables.size(); i++) {
//查詢表內數據量
int num = countByTableName(connection, p, tables.get(i));
//finalNum為邊界值，此處為1000萬
if (num > finalNum) {
……
}
……
}
3.根據所有的表計算現有的虛擬節點
ConsistentHash<String> consistentHashOld = getConsistentHash(tables);
4.把表加倍
List<String> tablesNew = deepCopy(tables); //注意一定要采用深復制
int tableSize = tablesNew.size();
for (int y = 0; y < tableSize; y++) {
String tableNameNew = tableName + (tablesNew.size() + 1);
//創建表
createTable(connection, p, d.toString(), tableNameNew);
tablesNew.add(tableNameNew);
tableDelete.add(tableNameNew);
}
5.計算加倍后的虛擬節點
ConsistentHash<String> consistentHashNew = getConsistentHash(tablesNew);
6.數據遷移
for (int z = 0; z < tableSize; z++) {
String tableNameOld = tablesNew.get(z);
//查詢試驗信息id不重復的試驗數據信息
List<String> disData = selectExperimentIdDis(connection, p, tableNameOld);
List<String> deleteList = new LinkedList<>();
for (String experimentId : disData) {
//如果數據hash計算 原所在表與新建表之后不一致，執行轉移
if (!consistentHashNew.get(experimentId).equals(consistentHashOld.get(experimentId))) {
//新增到新表數據
insertHash(connection, p, experimentId, consistentHashOld.get(experimentId),
consistentHashNew.get(experimentId));
//刪除數據集合
deleteList.add(experimentId);
//刪除舊表數據
final int defaultDelNum = 1000;
if (deleteList.size() == defaultDelNum) {
deleteInbatch(connection, p, deleteList, tableNameOld);
deleteList.clear();
}
}
}
//刪除舊表數據
if (deleteList.size() > 0) {
deleteInbatch(connection, p, deleteList, tableNameOld);
}
}
總結
以上為我們所做的一致性hash實踐，其中還存在很多問題，比如遷移過程單線程導致遷移較慢、自動擴容機制不智能、遷移過程中數據訪問不穩定等情況。
我們將會在后續的開發中逐步進行完善改進。
以上就是我們針對一致性hash在oracle分表中的實踐