Apache Ignite 學習筆記(五): Primary和backup數據同步模式和處理分片丟失的策略

時間 2019-12-08

標籤 apache ignite 學習筆記 primary backup 數據同步模式處理分片丟失策略欄目 Apache 简体版

原文原文鏈接

上一篇文章咱們介紹了Ignite數據網格中不一樣的數據分片冗餘策略：Replicated和Partition模式。不管是哪一種模式，其實就是經過對數據分片在不一樣的節點上作多個拷貝來保證數據的可用性。在一個多個節點組成的分佈式系統中，一旦須要作數據拷貝，天然就要考慮數據拷貝的過程是同步的仍是異步的。並且，在partition模式下，一個節點也許不會有數據的全部分片，那勢必會出現某個數據分片的primary和backup拷貝因爲節點故障，在集羣中訪問不到的狀況。這篇文章咱們就接着看看，針對數據拷貝以及數據分片丟失，Ignite提供了哪些選項，咱們又該怎樣處理。java

Primary和Backup之間的同步/異步拷貝

Ignite針對primary和backup之間的數據拷貝提供了三種同步模式：node

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由於是徹底異步的，因此有可能會出現數據丟失，這裏犧牲了數據的可用性，換取更好的寫性能。git

咱們能夠經過XML配置文件或者是代碼中之間配置同步模式。下面是XML配置文件：spring

<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代碼中設置同步模式apache

...
    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的事件：maven

啓動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節點的代碼：tcp

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查詢，還提供了哪些功能強大的查詢方式。