Apache Ignite 學習筆記(五): Primary和backup資料同步模式和處理分片丟失的策略
阿新 • • 發佈：2019-06-13
上一篇文章我們介紹了Ignite資料網格中不同的資料分片冗餘策略：Replicated和Partition模式。無論是哪種模式，其實就是通過對資料分片在不同的節點上做多個拷貝來保證資料的可用性。在一個多個節點組成的分散式系統中，一旦需要做資料拷貝，自然就要考慮資料拷貝的過程是同步的還是非同步的。而且，在partition模式下，一個節點也許不會有資料的所有分片，那勢必會出現某個資料分片的primary和backup拷貝由於節點故障，在叢集中訪問不到的情況。這篇文章我們就接著看看，針對資料拷貝以及資料分片丟失，Ignite提供了哪些選項，我們又該怎樣處理。
Primary和Backup之間的同步/非同步拷貝
Ignite針對primary和backup之間的資料拷貝提供了三種同步模式：
- PRIMARY_SYNC: 預設情況下Ignite採用的同步模式。寫cache的操作在資料分片的primary節點成功寫入即可返回，不用等待backup節點資料成功寫入。這也意味著，如果此時從backup節點讀資料，有可能讀到的任然是舊資料。
- FULL_SYNC： 寫cache的操作在primary節點和backup節點都成功寫入後返回。和PRIMARY_SYNC模式相比，這個模式保證了寫入成功後節點之間的資料都一樣。
- FULL_ASYNC： 寫cache的操作不用等primary節點和backup節點成功寫入即可返回。和PRIMARY_SYNC模式相比，此時即便是讀primary節點的資料都有可能讀到舊資料。
三種同步模式如何選擇，完全取決於應用對資料一致性，可用性和效能的要求。FULL_SYNC保證新的資料同步到了primary和backup節點上，自然對寫操作的效能影響是最大的。PRIMARY_SYNC則只保證資料同步到了primary節點上，這個模式犧牲一定的可用性換取了比FULL_SYNC更好的寫效能。而FULL_ASYNC因為是完全非同步的，所以有可能會出現資料丟失，這裡犧牲了資料的可用性，換取更好的寫效能。
我們可以通過XML配置檔案或者是程式碼中之間配置同步模式。下面是XML配置檔案：
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd"> <bean id="grid.cfg" class="org.apache.ignite.configuration.IgniteConfiguration"> <property name="discoverySpi"> <bean class="org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi"> <property name="ipFinder"> <bean class="org.apache.ignite.spi.discovery.tcp.ipfinder.multicast.TcpDiscoveryMulticastIpFinder"> <property name="multicastGroup" value="224.0.0.251"/> </bean> </property> </bean> </property> <property name="cacheConfiguration"> <bean class="org.apache.ignite.configuration.CacheConfiguration"> <!-- 設定快取名字. --> <property name="name" value="TEST"/> <!-- 設定快取模式. --> <property name="cacheMode" value="PARTITIONED"/> <property name="backups" value="1"/> <!-- 下面將快取設定為replicated模式 --> <!--property name="cacheMode" value="REPLICATED"/--> <property name="writeSynchronizationMode" value="FULL_SYNC"/> </bean> </property> </bean> </beans>
下面例子是如何在Java程式碼中設定同步模式
...
CacheConfiguration<String, String> cacheCfg = new CacheConfiguration("TEST");
cacheCfg.setCacheMode(CacheMode.PARTITIONED);
cacheCfg.setBackups(1);
cacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
...資料分片丟失的處理
在partition模式下， 資料分片後存放在primary和backup節點上，一旦出現某塊資料分片的所有primary和backup拷貝由於節點故障無法訪問時，就出現了“partition loss"的情況。用上一篇文章的partitoned cached圖來舉個例子：
圖中cache用的模式partition模式，backup數量是1，所以資料分片有一個primar和backup拷貝，如果JVM1和JVM4出現故障，那麼分片D的primary拷貝和backup拷貝全都無法訪問。這時候，如果允許使用者的讀寫操作繼續讀取分片D資料，那資料的一致性就無法保證了。我們可以通過監聽EVT_CACHE_REBALANCE_PART_DATA_LOST事件，及時知道叢集中出現partition loss，然後採取相應措施。另外，Ignite提供了不同的處理策略，讓你可以針對不同的場景選擇不同的策略：
- IGNORE: 如果不進行配置，這是預設情況下的策略。即使出現了partition loss的情況，Ignite會自動忽略並且會清空和partion loss相關的狀態不會觸發EVT_CACHE_REBALANCE_PART_DATA_LOST事件。
- READ_WRITE_ALL: Ignite允許所有的讀寫操作，就好像partition loss沒發生過。和IGNORE策略最大的不同，該策略雖然允許繼續讀寫，但會觸發EVT_CACHE_REBALANCE_PART_DATA_LOST事件。
- READ_WRITE_SAFE: 允許對沒有丟失的partition的讀寫操作，但是對已經丟失的partition的讀寫操作會失敗並拋異常。
- READ_ONLY_ALL: 允許對丟失的和正常的partition的讀操作，但是寫操作會失敗並拋異常。
- READ_ONLY_SAFE: 所有的寫操作和對丟失partition的讀操作都會失敗並拋異常。允許對正常的partition的讀操作。
下面，讓我們用一個例子演示下如何配置partition loss的策略，以及如何通過監聽EVT_CACHE_REBALANCE_PART_DATA_LOST處理paritition loss的事件：
- 啟動2個server例項。
- server例項啟動後，啟動一個client節點連上叢集，用CacheMode.PARTITIONED模式建立一個backup數量為0的cache(將backup數量設為0為了方便模擬partition丟失的場景)， 然後往cache裡寫一些資料，並監聽EVT_CACHE_REBALANCE_PART_DATA_LOST事件。
- 關掉一個server例項模擬節點故障觸發partition loss。
- 觀察client例項能否收到EVT_CACHE_REBALANCE_PART_DATA_LOST事件，在發生partition loss後啟用不同策略繼續讀寫cache的行為，以及如何重置叢集狀態讓讀寫恢復正常。
因為server節點的邏輯很簡單(實際上2個server節點就是啟動後組成一個Ignite叢集)，我們看看client節點的程式碼：
public class IgnitePartitionLossExampleClient {
private static AtomicBoolean partitionLost = new AtomicBoolean(false);
public static void main(String[] args) {
Ignite ignite;
if (args.length == 1 && !args[0].isEmpty()) {
//如果啟動時指定了配置檔案,則用指定的配置檔案
System.out.println("Use " + args[0] + " to start.");
ignite = Ignition.start(args[0]);
} else {
//如果啟動時沒指定配置檔案,則生成一個配置檔案
System.out.println("Create an IgniteConfiguration to start.");
TcpDiscoverySpi spi = new TcpDiscoverySpi();
TcpDiscoveryMulticastIpFinder ipFinder = new TcpDiscoveryMulticastIpFinder();
ipFinder.setMulticastGroup("224.0.0.251");
spi.setIpFinder(ipFinder);
IgniteConfiguration cfg = new IgniteConfiguration();
cfg.setDiscoverySpi(spi);
cfg.setClientMode(true);
//預設由於效能原因，Ignite會忽略所有事件，這裡要主動配置需要監聽的事件
cfg.setIncludeEventTypes(EventType.EVT_CACHE_REBALANCE_PART_DATA_LOST);
ignite = Ignition.start(cfg);
}
// 建立一個TEST快取, cache mode設為PARTITIONED, backup數量為1, 並把partition loss policy設為READ_WRITE_SAFE
CacheConfiguration<String, String> cacheCfg = new CacheConfiguration<>();
cacheCfg.setName("TEST");
cacheCfg.setCacheMode(CacheMode.PARTITIONED);
cacheCfg.setBackups(0);
cacheCfg.setPartitionLossPolicy(PartitionLossPolicy.READ_WRITE_SAFE);
IgniteCache<String, String> cityProvinceCache = ignite.getOrCreateCache(cacheCfg);
// Local listener that listens to local events.
IgnitePredicate<CacheRebalancingEvent> locLsnr = evt -> {
try {
System.out.println("=========Received event [evt=" + evt.name() + "]==========");
Collection<Integer> lostPartitions = cityProvinceCache.lostPartitions();
if (lostPartitions != null) {
partitionLost.set(true);
}
return true; // Continue listening.
} catch (Exception e) {
System.out.println(e);
}
System.out.println("=========Stop listening==========");
return false;
};
// Subscribe to specified cache events occuring on local node.
ignite.events().localListen(locLsnr,
EventType.EVT_CACHE_REBALANCE_PART_DATA_LOST);
List<String> cities = new ArrayList<String>(Arrays.asList("Edmonton",
"Calgary", "Markham", "Toronto", "Richmond Hill", "Montreal"));
// 寫入一些資料, key是城市的名字,value是省的名字
populateCityProvinceData(cityProvinceCache);
//用下面的while迴圈不停模擬對cache的讀寫操作
while(true) {
try {
for (String city : cities) {
try {
if (!partitionLost.get()) {
//如果cache一切正常，則正常讀
getAndPrintCityProvince(city, cityProvinceCache);
} else {
//如果cache出現partition lost，模擬錯誤處理, 我們這裡簡單把cache
//lost partiton重置，並重新寫入資料
Collection<Integer> lostPartitions = cityProvinceCache.lostPartitions();
System.out.println("Cache lost partitions: " + lostPartitions.toString());
ignite.resetLostPartitions(Arrays.asList("TEST"));
populateCityProvinceData(cityProvinceCache);
partitionLost.set(false);
}
} catch(CacheException e) {
e.printStackTrace();
}
}
Thread.sleep(1000);
}
catch (Exception e) {
e.printStackTrace();
}
}
}
private static void populateCityProvinceData(IgniteCache<String, String> cityProvinceCache) {
System.out.println("Populate city province data！");
cityProvinceCache.put("Edmonton", "Alberta");
cityProvinceCache.put("Calgary", "Alberta");
cityProvinceCache.put("Markham", "Ontario");
cityProvinceCache.put("Toronto", "Ontario");
cityProvinceCache.put("Richmond Hill", "Ontario");
cityProvinceCache.put("Montreal", "Quebec");
}
private static void getAndPrintCityProvince(String city, IgniteCache<String, String> cityProvinceCache) {
System.out.println(city + " is in " + cityProvinceCache.get(city));
}
}- 在第30~31行，我們將backup數量設為0並呼叫setPartitionLossPolicy函式將cache在partition丟失後的模式改為READ_WRITE_SAFE，這種模式下允許對沒有丟失的partition的讀寫操作，但是對已經丟失的partition的讀寫操作會失敗並拋異常。
- 在35~52行，我們建立了一個對CacheRebalancingEvent的監聽器，並且通過localListen函式將監聽器註冊給Ignite，這樣在Ignite叢集中一旦出現EVT_CACHE_REBALANCE_PART_DATA_LOST事件，該監聽器就會被呼叫。在監聽器中，我們不但列印了觸發該listener的事件，還通過cache的lostPartitions函式返回丟失掉的partition的資訊，如果確實有partition丟失了，listener還會把partitionLost置為true用來觸發第70~71行的修復程式碼。
- 在註冊監聽器時有一點需要注意，Ignite為了效能，預設會忽略對所有事件的通知，為了能得到EVT_CACHE_REBALANCE_PART_DATA_LOST的事件通知，我們需要在啟動Ignite節點時，顯式的開啟我們關心的事件通知(程式碼第22行，也可以通過xml檔案配置，具體見例項程式碼裡的ignite-cache.xml配置檔案)。
- 第59行，我們往cache中寫入一些初始資料。接著在62~88行的一個while迴圈中，我們不斷的從cache裡讀資料。在每次讀cache前，我們都檢查一些partitionLost是否有被置為true。如果沒有，我們就讀cache並打印出來(68行)；否則，說明cache的某些partition丟失了，我們通過lostPartitions函式得到丟失的partition的資訊，並打印出來。生產環境中，partition丟失代表著資料丟失，這時需要從外部幫忙恢復資料，或者檢查下丟失的資料是否重要，然後才能將cache恢復正常讀寫。在例子中，由於我們知道全部資料集，所以可以直接呼叫resetLostPartitions將cache恢復，並且重寫一遍資料，這樣後面的讀操作都能成功。
在關掉一個server節點後，client節點會在console列印如下結果：
=========Received event [evt=CACHE_REBALANCE_PART_DATA_LOST]==========
Cache lost partitions: [0, 3, 6, 7, 8, 9, 10, 11, 12, 19, 22, 23, 24, 27, 32, 34, 36, 37, 38, 39, 40, 42, 44, 50, 51, 52, 53, 56, 57, 59, 61, 64, 66, 69, 70, 71, 76, 78, 79, 80, 82, 83, 86, 87, 88, 91, 92, 95, 97, 98, 100, 101, 103, 106, 108, 112, 115, 117, 120, 121, 123, 124, 125, 128, 129, 135, 137, 138, 140, 141, 144, 145, 148, 149, 150, 151, 155, 157, 161, 162, 165, 166, 167, 169, 170, 171, 172, 179, 181, 183, 185, 188, 190, 191, 200, 201, 202, 204, 205, 212, 213, 214, 216, 217, 222, 223, 225, 226, 230, 231, 233, 235, 238, 239, 240, 242, 243, 244, 245, 246, 247, 255, 256, 257, 259, 261, 262, 263, 264, 266, 268, 271, 273, 274, 278, 281, 282, 284, 285, 288, 290, 291, 293, 296, 297, 299, 302, 306, 307, 312, 314, 316, 317, 318, 319, 322, 325, 326, 330, 333, 335, 336, 337, 339, 340, 343, 344, 345, 346, 348, 349, 351, 352, 354, 356, 359, 360, 362, 365, 367, 369, 370, 371, 373, 374, 379, 381, 383, 385, 386, 387, 391, 396, 398, 401, 403, 404, 405, 407, 410, 412, 413, 423, 425, 426, 427, 431, 436, 437, 438, 439, 440, 441, 442, 443, 444, 445, 446, 447, 448, 451, 453, 456, 457, 462, 465, 467, 469, 471, 472, 473, 476, 477, 479, 481, 482, 483, 484, 485, 486, 487, 490, 492, 499, 500, 501, 502, 503, 504, 505, 507, 509, 511, 519, 520, 522, 523, 524, 526, 527, 528, 529, 531, 536, 538, 540, 541, 542, 544, 546, 547, 548, 549, 552, 553, 554, 555, 558, 559, 563, 565, 567, 570, 572, 573, 577, 578, 580, 581, 585, 586, 589, 590, 591, 593, 600, 601, 602, 603, 604, 605, 606, 607, 608, 610, 611, 616, 617, 619, 621, 624, 627, 628, 629, 630, 631, 635, 637, 639, 643, 646, 648, 652, 653, 658, 661, 665, 667, 670, 673, 675, 686, 687, 691, 693, 695, 696, 700, 703, 705, 707, 711, 712, 716, 718, 719, 720, 721, 722, 724, 730, 731, 732, 733, 734, 735, 736, 737, 738, 739, 742, 743, 744, 745, 750, 751, 752, 756, 758, 760, 763, 764, 766, 769, 775, 776, 777, 778, 779, 780, 782, 786, 789, 790, 792, 793, 794, 797, 799, 801, 802, 808, 809, 810, 812, 813, 814, 815, 819, 820, 821, 822, 824, 825, 827, 830, 831, 834, 835, 836, 837, 838, 843, 844, 846, 847, 848, 852, 854, 859, 863, 864, 866, 867, 874, 876, 878, 879, 881, 882, 883, 884, 885, 888, 891, 895, 896, 897, 899, 900, 903, 904, 905, 907, 908, 910, 911, 915, 916, 917, 920, 922, 924, 928, 933, 935, 936, 940, 942, 943, 945, 948, 949, 950, 951, 952, 955, 956, 960, 962, 965, 969, 971, 972, 973, 979, 983, 985, 990, 994, 995, 1001, 1003, 1010, 1012, 1013, 1014, 1015, 1017, 1018, 1019, 1020, 1021, 1022, 1023]
Populate city province data！這表示client節點的確收到EVT_CACHE_REBALANCE_PART_DATA_LOST的事件通知，lostPartitions函式的返回結果也證明了哪些partition丟失了。當我們把第31行程式碼註釋掉後(相當於我們把cache的lost partition policy設為了預設的IGNORE)，重新編譯再跑同樣的試驗，我們會發現這時client節點不會收到partition lost的相關事件。
總結
這篇文章我們介紹了Ignite提供的primary和backup資料之間三種同步模式，以及這三種同步模式的適用場景。我們還介紹了在出現partition丟失情況下，Ignite提供的事件通知機制。通過一個簡單的例子和程式碼，我們展示瞭如何新增監聽Ignite的事件，並處理partition丟失事件。這篇文章裡用到的例子的完整程式碼和maven工程可以在這裡找到。 Client對應的xml配置檔案在src/main/resources目錄下。
下一篇，我們將繼續深入瞭解一下Ignite cache除了簡單的key/value查詢，還提供了哪些功能強大的查詢方式