Hazelcast集群服务(4)——分布式Map

本篇将介绍Hazelcast实现的分布式Map的原理和使用方法。

分布式Map基础功能

    Map是我们再最常用的数据接口之一，时常用于存储某种关系值。在前面介绍Hazelcast的文章中已经用Map举了很多分布式环境使用的例子。下面我们将由浅入深的介绍Hazelcast的Map。

    在Hazelcast中自定义了一个名为IMap的接口，该接口自java.util.concurrent.ConcurrentMap接口，所以可以通过常规的Map::get和Map::put方法来控制集群Map的数据。先看一个Map使用简单的例子：

    （再次申明，文中所有的例子的代码都在：https://github.com/chkui/hazelcast-demo，git clone到本地用Maven导入即可运行）

    首先创建一个服务端节点，并向节点中的Map添加数据。

// 
// 服务端节点
public class ServiceNode {
    public static void main(String[] args) {
        // 获取Hazelcast实例
        HazelcastInstance ins = Hazelcast.newHazelcastInstance();

        // 从集群中读取Map实例
        Map<Integer, String> map = ins.getMap("default map");

        // 向集群中添加数据
        System.out.println("Begin insert data");
        map.put(1, "Cait Cassiopeia");
        map.put(2, "Annie");
        map.put(3, "Evelynn");
        map.put(4, "Ashe");
        System.out.println("End");
        
    }
}

    然后创建一个客户端节点，从节点的Map读取数据。

//
// 客户端节点
public class ClientNode {
    public static void main(String[] args) {
        // 获取Hazelcast实例
        HazelcastInstance ins = Hazelcast.newHazelcastInstance();

        // 从集群中读取Map实例
        Map<Integer, String> map = ins.getMap("default map");

        // 输出map中数据
        map.forEach((k,v)->{
            System.out.println("Pos:" + k + ". name:" + v);
        });
        
    }
}

    这就是使用集群Map的过程，和常规Map并没有什么差异。在使用集群Map时，最主要是了解Map的各种配置对Map功能的影响，以及Hazelcast为Map提供了哪些扩展接口。下面将会结合配置文档，说明每一个配置参数的功效。

    先看分布式Map的基础配置参数：

<map name="default">
    <backup-count>0</backup-count>
    <async-backup-count>1</async-backup-count>
    <read-backup-data>true</read-backup-data>
    <in-memory-format>BINARY</in-memory-format>
    <eviction-policy>LRU</eviction-policy>
    <time-to-live-seconds>0</time-to-live-seconds>
    <max-idle-seconds>0</max-idle-seconds>
    <min-eviction-check-millis>150</min-eviction-check-millis>
    <max-size policy="PER_NODE">5000</max-size>
    <eviction-percentage>25</eviction-percentage>
</map>

    下面这个是一个使用代码配置的例子：

//
// 代码设置Map
public class StaticMapConfig {
    public static void main(String[] args) {
        MapConfig mapConfig = new MapConfig();
        mapConfig.setName("cacheMap")// 设置Map名称
                .setInMemoryFormat(InMemoryFormat.BINARY)// 设置内存格式
                .setBackupCount(1);// 设置副本个数

        mapConfig.getMapStoreConfig()//
                .setWriteDelaySeconds(60)//
                .setWriteBatchSize(1000);// 设置缓存格式

        mapConfig.addMapIndexConfig(new MapIndexConfig().setAttribute("id").setOrdered(true));// 增加索引
        mapConfig.addMapIndexConfig(new MapIndexConfig().setAttribute("name").setOrdered(true));
    }
}

    下面我们将一一介绍每个配置的含义。

backup-count

   备份副本个数[0~Integer.MAX_VALUE]。 前面的博文已经介绍，集群中分布式存储的数据都会被均匀的存储在每个节点上。我们使用Map进行分布式数据存储时，每个节点会按条目（Entry）数将数据进行分布，并且每条数据都会有备份。例如集群中的一个Map有1000条数据，此时有2个节点，那么每个节点会存储1000条数——500条主数据和500条备份数据，以此类推，当有5个节点时，每个节点200条主数据加200条备份数据。

    backup-count 就是用来定义备份副本个数的，默认为1。当设置为0时，集群中不会有任何数据副本。这个参数需要根据数据的业务需要来定义，值越大，需要备份的副本就越多，集群中需要处理的数据就越多，会导致性能降低。

async-backup-count

    异步备份副本的个数[0~Integer.MAX_VALUE]。这个参数和backup-count类似，也是指定备份副本的个数，区别在于这里指定的副本，是异步备份的。例如，我们执行map.put(key,value)时，Hazelcast会先向主表添加数据。这时如果指定了backup-count = 1，会先更新副本数据，然后再return到调用put方法的线程，在添加数据完成之前，调用线程都是被阻塞的。如果指定了async-backup-count = 1，那么当添加主表数据成功后，会直接返回给调用线程，然后再去异步执行数据备份。使用同步方法还是异步方法，需要根据业务数据的重要性来决定，如果是一定不能丢失的数据，最好用同步方法，如果备份出现异常，会马上通知到调用线程然后执行补偿操作。

read-backup-data

    副本直读数据[true|false]。当我们的集群中有一个map的备份数据后，这些备份数据也是分散存储在各个节点的。当某个节点需要读取数据时，read-backup-data设置为false表示只能从主表数据读取，设置为true表示可以从备份副本中读取数据。设置为true，可以提升数据的读取数据，因为在某个节点要读取某条数据时，该节点正好有该条数据，可以减少网络交互消耗。但是设置为trure可能会导致“数据脏读”。

in-memory-format

    内存数据格式[BINARY|OBJECT]。

    BINARY：这是默认配置。数据将被序列化成二进制的方式存储。如果在应用中Map的主要执行的都是像get和put这样的常规操作，建议使用这个配置。

    OBJECT：数据将以非序列化的结构存储。这个配置有利于当Map中存储的对象比较复杂，对其进行序列化的成本较高时。当需要对存储复杂对象的Map条目进行大量查询时，建议使用OBJECT。

    用一个场景来说明他们的区别。我们的对象都是存储在每个节点中的，当某个节点需要get不在本地的一条数据时，Hazelcast需要去其他节点获取数据。此时如果以二进制的方式存储，不用进行序列化，直接将数据进行传输，而如果以对象的方式存储，在传输之前，需要进行一次序列化操作，然后再传递数据。

eviction-policy

    数据释放策略[NONE|LRU|LFU]。这是Map作为缓存的一个参数，用于指定数据的回收算法。默认为NONE。

    NONE：当设置为NONE时，不会发生数据回收，同时max-size会失效。但是仍然可以使用time-to-live-seconds和max-idle-seconds参数来控制数据留存时间。

    LRU：“最近最少使用“策略。

    LFU：“最不常用的使用”策略。

time-to-live-seconds（TTL）

    数据留存时间[0~Integer.MAX_VALUE]。缓存相关参数，单位秒，默认为0。这个参数决定了一条数据在map中的停留时间。当数据在Map中留存超过这个时间并且没有被更新时，它会根据指定的回收策略从Map中移除。值为0时，意味着无求大。

max-idle-seconds

    数据的最大空闲时间[0~Integer.MAX_VALUE]。缓存相关参数，单位秒，默认值为0。当条目的空闲时间大于这个数值时，将会被自动释放。条目的空闲是指没get、put、EntryProcessor.process或containsKey方法被调用。默认值为0，意味着无求大。

min-eviction-check-millis

    分区数据释放检查周期[0~Integer.MAX_VALUE]。缓存相关参数，单位秒，默认值为100。前面提到了Hazelcast会对map存储的数据进行释放。为了移除这些数据，有一个轮询工作在不间断的执行。换一种说法，就是数据释放的频率。当设置为0时，每一次数据的put操作，都会导致一次数据释放执行。

max-size

    Map中存储条目的最大值[0~Integer.MAX_VALUE]。默认值为0。当条目数量接近最大值时，map将基于配置的策略进行数据释放。如果期望max-size生效，必须eviction-policy将设置为NONE之外的其他值。max-size中包含一个属性参数——policy，他定义了max-size对应的存储策略，回收机制会根据这个策略检查数据。其值有[PER_NODE|PER_PARTITION|USED_HEAP_SIZE|USED_HEAP_PERCENTAGE|FREE_HEAP_SIZE|FREE_HEAP_PERCENTAGE]。

    PER_NODE：max-size指定单个集群成员中map条目的最大数量。这是max-size的默认策略。如果使用这个配置，需要注意max-size的值必须大于分区的数量（默认为271）。

    PER_PARTITION：max-size指定每个分区存储的map条目最大数。这个策略建议不要在小规模的集群中使用，因为小规模的集群，单个节点包含了大量的分区，在执行回收策略时，会去按照分区的划分组个检查回收条件，导致效率低下。

    USED_HEAP_SIZE：指在每个Hazelcast实例中，max-size指定map所占用的内存堆的（以megabytes计算，兆字节）最大值。需要注意这个策略不能工作在in-memory-format=OBJECT，因为当数据被设置为OBJECT时，无法确定所占用的内存大小。

    USED_HEAP_PERCENTAGE：每个Hazelcast实例中，max-size指定map占用内存堆的百分比。例如，JVM被设置有1000MB，而这个值设置为max-size=10，当map条目数占用的堆数据超过100MB时，Hazelcast开始执行数据释放工作。需要注意的是当使用这个策略时，不能将in-memory-format设置为OBJECT，理由同上。

    FREE_HEAP_SIZE：max-size指定了单个JVM的堆最小空闲空间，单位为megabytes。

    FREE_HEAP_PERCENTAGE：max-size指定单个JVM的最小空闲空间的百分比。例如JVM分配了1000MB的空间，这个值设置为10，当空闲堆只有100MB时，会引发map的数据清除放行为。

eviction-percentage

    数据清理的百分比[0-100]。当触发数据清除条件，这个参数所配置的百分比MAP条目将被释放。例如设置为25，25%的条目将会被清除。将这个值设置较小时会导致Map中只有较少的条目被释放，导致Hazelcast频繁的执行数据清除操作。如果map的条目数据经常被添加，请将这个比率提高，默认为25。

Near Cache

    Near cache是Hazelcast分布式Map重要的功能之一。根据前面的知识我们知道，Hazelcast的所有数据都是按照分区存储在每个集群节点之上的。假设集群中的一个节点需要根据key读取某条数据，而这些数据被放置在其他的节点。这样每次Map.get操作都会导致一次网络数据传输，如果节点分布较广、传输能力参差不齐，会导致大量的网络拥塞，进而影响每个节点的执行。尤其是某个map的读操作远远多于写操作时，我们可以考虑使用Near cache功能。Near cache会将那些被某个节点经常使用的数据存储到当前节点或“附近”节点，以此来减少过多的网络传输工作。使用Near cache也会导致一些问题出现，在使用之前，必须了解一下问题：

1. 使用Near cache功能会导致集群中的成员额外存储缓存数据，会增加内存的消耗。
2. Near cache会破坏数据一致性性，可能会出现“脏读”现象，因此在频繁写或数据一致性要求较高的应用中不建议使用。
3. 建议在高频读操作的Map中启用Near cache功能，这样可以极大的提升执行效率。

    Near cache的配置都在near-cache元素中。下面介绍Near cache的相关参数。

<map name="my-read-mostly-map">
    <near-cache name="default">
        <in-memory-format>BINARY</in-memory-format>
        <max-size>5000</max-size>
        <time-to-live-seconds>0</time-to-live-seconds>
        <max-idle-seconds>60</max-idle-seconds>
        <eviction-policy>LRU</eviction-policy>
        <invalidate-on-change>true</invalidate-on-change>
        <cache-local-entries>false</cache-local-entries>
    </near-cache>
</map>

in-memory-format

    与Map的in-memory-format配置一样，指定了Map在Near cache中的存储格式。参见前文介绍的in-memory-format功能。

max-size

    Near cache缓存中存储的最大条目数[0~Integer.MAX_VALUE]。Near cache会根据eviction-policy指定的策略来释放数据。默认为0，表示不限定最大条目数。

time-to-live-seconds

    单条数据在Near cache中的最大驻留时间[0~Integer.MAX_VALUE]。单位秒，默认为0。如果存储在Near cache中的某条数据在Near cache中的驻留时间（没有被更新）超过这个时间，则在执行数据回收时会被释放掉。值为0时表示永远不会过期。

max-idle-seconds

    单条数据在Near cache中的最大失效时间[0~Integer.MAX_VALUE]。单位秒，默认值为0。如果存储在Near cache中的某条数据在指定时间内没有被读取，则认为该条数据失效。此时在执行数据回收时会释放掉该条数据。值为0时表示用于不会失效。

eviction-policy

    数据释放策略，见前面 Map释放策略 的说明。

invalidate-on-change

    设定当Near cache中的某条数据被更新或移除时，是否对其执行释放[true|false]。默认为true。

cache-local-entries

    指定那些已经被存储在当前节点的数据条目，是否也进行Near cache缓存[true|false]。这个参数最大的作用在于，可以将Near cache的内存格式设定成和Map存储格式不一样的方式。默认为fasle。  

MapStore数据持久化

    后续的篇幅将介绍Hazelcast分布式Map的一些基础功能。这里先介绍如何对数据库进行数据读写。Hazelcast分布式Map的持久化数据读写通过MapStore来实现。请看下面这个例子：

    先是一个配置文件，后面在说他的意义：

<!--https://github.com/chkui/hazelcast-demo/blob/master/src/main/java/org/palm/hazelcast/map/store/mapStoreConfig.xml -->
<?xml version="1.0" encoding="UTF-8"?>
<hazelcast
    xsi:schemaLocation="http://www.hazelcast.com/schema/config hazelcast-config-3.6.xsd"
    xmlns="http://www.hazelcast.com/schema/config" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    <map name="demo">
        <map-store enabled="true" initial-mode="EAGER">
            <class-name>org.palm.hazelcast.map.store.MapStoreExample</class-name>
            <write-delay-seconds>60</write-delay-seconds>
            <write-batch-size>1000</write-batch-size>
            <write-coalescing>true</write-coalescing>
        </map-store>
    </map>
</hazelcast>

然后定义一个当Map发生数据读写时对数据库进行操作的MapStore类：

//https://github.com/chkui/hazelcast-demo/blob/master/src/main/java/org/palm/hazelcast/map/store/MapStoreExample.java
public class MapStoreExample implements MapStore<Integer, String> {
    Map<Integer, String> store;
    public MapStoreExample(){
        store = new HashMap<Integer, String>();
        store.put(1, "Azeroth");
        store.put(2, "Duskwood");
        store.put(3, "Elwynn Forest");
        store.put(4, "Deadwind Pass");
        store.put(5, "Dead Mines");
        store.put(6, "Grand Hamlet");
        store.put(7, "Dark Portal");
        store.put(8, "Ashenvale");
        store.put(9, "Felwood");
        store.put(10, "Orgrimmar");
    }
    @Override
    public String load(Integer key) {//读取
        if(store.size() < key){
            key = 0;
        }
        return store.get(key);
    }
    @Override
    public Map<Integer, String> loadAll(Collection<Integer> arg0) {//读取所有
        return store;
    }
    @Override
    public Iterable<Integer> loadAllKeys() {// 读取所有键值
        return store.keySet();
    }
    @Override
    public void delete(Integer key) {// 删除键值
        store.remove(key);
    }
    @Override
    public void deleteAll(Collection<Integer> list) {// 删除所有键值
        list.forEach(key->store.remove(key));
    }
    @Override
    public void store(Integer key, String value) {// 存储键值
        store.put(key, value);
    }
    @Override
    public void storeAll(Map<Integer, String> map) {// 存储所有键值
        store.putAll(map);
    }
}

   最后时一个测试main：

//https://github.com/chkui/hazelcast-demo/blob/master/src/main/java/org/palm/hazelcast/map/store/MapStoreExampleMain.java
public class MapStoreExampleMain {
    public static void main(String[] args) throws FileNotFoundException {
        //加载配置
        Config config = new ClasspathXmlConfig("org/palm/hazelcast/map/store/mapStoreConfig.xml");
        //创建Hazelcast实力
        HazelcastInstance ins = Hazelcast.newHazelcastInstance(config);
        //获取Map
        Map<Integer, String> map = ins.getMap("demo");
        println(map.get(1));//输出第一条数据
        
        map.put(11, "Moonbrook");//添加一条数据
        println(map.get(11));//输出第一条数据
        
        map.remove(11);//移除添加的数据
        println(map.get(11));//输出被移除的数据
    }
    static private void println(Object obj){
        System.out.println(obj);
    }
}

     仔细看代码例子的兄弟应该明白怎么回事了吧。配置文件中<map-store>元素定义了Mapstore的行为，<class-name>定义了当发生数据读写时要调用的存储类，该类需要实现MapStore接口。MapStore接口定义了当对Map进行put、get、remove操作时会被调用实现类的store、load、delete方法，我们可以通过自己的代码来完成对数据库的写入和读取操作。

    MapStore支持Read-Through、Write-Through、Write-Behind模式，下面简单介绍一下这三种模式：

　1.Read-Through：当应用系统向缓存系统请求数据时（例如使用key=x向缓存请求数据）；如果缓存中并没有对应的数据存在（key=x的value不存在），缓存系统将向底层数据源的读取数据。如果数据在缓存中存在（命中key=x），则直接返回缓存中存在的数据。这就是所谓的Read-throug。

　2.Write-Through：当应用系统对缓存中的数据进行更新时（例如调用put方法更新或添加条目），缓存系统会同步更新缓存数据和底层数据源。

   3.Write-Behind：当应用系统对缓存中的数据进行更新时（例如调用put方法更新或添加条目），缓存系统会在指定的时间后向底层数据源更新数据。

write-delay-seconds

    我们可以使用<write-delay-seconds>来指定Map是使用Write-Through模式还是Write-Behind模式。

   当设定<write-delay-seconds>为0时，表示当执行Map::put时立刻调用注册的Mapstore的store方法，直到自定义的代码执行完毕返回后，Map::put方法才会返回，整个过程都会阻塞线程。这样就实现了Write-Through模式。

    当设置<write-delay-seconds>大于0时，表示延迟指定的时间后（秒）再异步调用Mapstore::store方法。整个过程不会产生阻塞，数据被添加到Map里后就返回给调用者。这就实现了Write-Behind模式。

    使用什么模式，请根据你的业务要求设置。下面是其他几个参数的含义。

write-batch-size

    批量更新参数[0~Integer.MAX_VALUE]。这个参数用于指定当累计多少次更新数据之后再调用Mapstore::store一次性写入数据库。例如设置为50，只有调用50次Map::put方法后，Hazelcast才会去调用一次storeAll，并且传入所有的更新数据。如果运行我上面提供的例子，你会发现MapStoreExample的store和remove方法并没有被调用。

write-coalescing

     标记是否执行所有更新[true|false]。默认为true。用于标记Mapstore::store是否获取所有的更新。在Write-Behind模式下，在延迟的时间中，可能已经对某个key对应的value值进行了多次更新，若参数设定为true，表示只传递最后一次更新给Mapstore::store(Mapstore::storeAll)。如果设置为false，会将所有的更新传递给Mapstore::storeAll。

enabled属性

    表示是否启用Mapstore[true|false]。

initial-mode属性

    初始化模式[LAZY|EAGER]。默认为LAZY，在此参数下，会对Mapstore执行异步初始化。当设置为“EAGER”时, 初始化过程将被阻塞，直到加载完成。

    除了上面的配置，我们可以通过Mapstore::loadAllKeys方法来设定当Map初始化时要加载的数据。如果loadAllKeys返回null，则不预加载任何数据。因此我们可以在loadAllKeys方法中指定当Map初始化时需要先加载的数据。

Map拦截器（Interceptors）

    我们可以为Map的所有方法添加拦截器，拦截器类似于AOP机制。对某些方法增加拦截器后，当这些方法被调用时，会根据拦截器的配置进入拦截器。拦截器的代码都是会组赛线程的，也就是说我们我们在拦截器中处理完某些事物后，必须马上返回。因为组赛了线程，拦截器的功能强大，可以对返回值、更新值进行任何修改。

    拦截器采取链式操作，也就是说可以为一个方法增加多个拦截器，这些拦截器会根据添加的次序逐个被调用。Hazelcast会根据拦截器的hashCode()方法来判断是否是同一个拦截器，如果是具有相同的hashcode，则认为是同一个拦截器，不会被添加。因此建议最好根据需要重载hashcode方法，以免重复添加相同的拦截器。

    下面是一个代码的例子，首先是实现了一个拦截器：

// https://github.com/chkui/hazelcast-demo/blob/master/src/main/java/org/palm/hazelcast/map/interceptor/InterceptorExample.java
public class InterceptorExample implements MapInterceptor, Serializable {
    private static final long serialVersionUID = -7591859317633144192L;
    @Override
     /**拦截get。可以返回其他值来替换原有值。返回null将不影响原有操作。 */
    public Object interceptGet(Object obj) {
        prinfln("get : " + obj);
        return obj;
    }

    @Override
     /**在get操作执行完毕后被调用。*/
    public void afterGet(Object obj) {
        prinfln("after get : " + obj);
    }

    @Override
    /**拦截put。返回的值会设置到map中。返回null时原有的put数据不会发生任何改变。 抛出异常会取消put操作。 */
    public Object interceptPut(Object oldValue, Object newValue ) {
        prinfln("put old value : " + oldValue);
        prinfln("put new value : " + newValue);
        return newValue;
    }

    @Override
     /**在put操作执行完毕后被调用。*/
    public void afterPut(Object obj) {
        prinfln("after put : " + obj);
    }
    @Override
    /**拦截remove。返回被删除对象或null将继续执行删除。 抛出异常会取消remove操作。 */
    public Object interceptRemove(Object obj) {
        prinfln("remove : " + obj);
        return null;
    }

    @Override
     /**在remove操作执行完毕后被调用。*/
    public void afterRemove(Object obj) {
        prinfln("afrer remove : " + obj);
    }

    private void prinfln(Object obj){
        System.out.println(obj);
    }
}

    Map拦截器的拦截器需要实现MapInterceptor接口。这些接口提供了针对get、put、remove实现了拦截。然后下面是一个使用的例子：

public class InterceptorDemo {
    public static void main(String[] args) {
        HazelcastInstance ins = Hazelcast.newHazelcastInstance();
        IMap<Integer, String> imap = ins.getMap("");
        imap.addInterceptor(new InterceptorExample());// 添加拦截器
        imap.put(1, "Mei");
        imap.put(1, "Tracer");
        imap.put(1, "D.va");
        imap.put(1, "Mercy");
        
        imap.get(1);
        imap.remove(1);
        System.out.println(imap.get(1));
    }
}

    我们使用IMap接口来获取map实例。然后使用IMap::addInterceptor方法来增加前面实现的拦截器。随后，所有针对这个Map的get、put、remove都会进入我们设定的拦截器。IMap::removeInterceptor可以用来移除一个拦截器。

Map事件监听器

    除了拦截器，Hazelcast还有监听器。监听器和拦截器的区别在于：拦截器会嵌入到业务流程中去，拦截器可以在处理数据的过程中改变数据和行为。而监听器并不会摄入到处理逻辑中，他只是观察到发生某个事件后，通知我们注册的监听器。下面还是同一个代码的例子说明监听器，先创建一个监听器：

// https://github.com/chkui/hazelcast-demo/blob/master/src/main/java/org/palm/hazelcast/map/listener/ListenerExample.java
public class ListenerExample implements EntryAddedListener<Integer, String>, EntryRemovedListener<Integer, String>, EntryUpdatedListener<Integer, String> {
    @Override
    public void entryUpdated(EntryEvent<Integer, String> entry) {//监听更新数据
        print("put entry. key = " + entry.getKey() + ". value = " + entry.getValue());
    }

    @Override
    public void entryRemoved(EntryEvent<Integer, String> entry) {//监听移除数据
        print("remove entry. key = " + entry.getKey() + ". value = " + entry.getValue());
    }

    @Override
    public void entryAdded(EntryEvent<Integer, String> entry) {//监听新增数据
        print("add entry. key = " + entry.getKey() + ". value = " + entry.getValue());
    }
    
    private void print(Object obj){
        System.out.println(obj);
    }
}

    一个监听器可以实现多个监听接口，除了例子中的接口，还有EntryEvictedListener（释放单条数据）、MapEvictedListener（清除Map数据）、MapClearedListener（清空Map数据）等。

    至此，Hazelcasl分布式Map的基本功能就介绍完毕了。

原文地址：https://my.oschina.net/chkui/blog/746496