【总结】Hazelcast之Distributed Map介绍

通过Hazelcast入门简介（http://angelbill3.iteye.com/blog/2342989），介绍了Hazelcast的基本信息，以及知道了Hazelcast能以多种数据结构存储，本文重点讲Map。
(注：本文基于Hazelcast 3.5.4版本写的)

文章有点长，考虑分成多篇文章。

目录：
1. Distributed Map的工作原理
2. Map的备份
    2.1 备份的一致性
    2.2 从备份中读取数据
3. Map的淘汰机制（eviction）
    3.1 Map淘汰机制的例子
    3.2 Map淘汰机制的配置
    3.3 淘汰机制配置的例子
    3.4 淘汰指定的元素
    3.5 淘汰所有的元素
4. 内存中的数据格式


文章太长另起一篇: TODO
5 Map持久化
6 Near Cache介绍
7 Map锁
8 Map统计
9 Map的监听 (listener)
10 拦截器
11 避免发生内存溢出异常



Hazelcast Map即IMap继承了java.util.concurrent.ConcurrentMap，由于Java的Concurrent接口继承java.util.Map，于是Imap也有了Map的特点, 我们可以像操作Java的Map（get/put）一样操作Hazelcast Map。此外，它跟ConcurrentMap一样，是线程安全的。

Java文档：http://docs.hazelcast.org/docs/3.5/javadoc/
源码：http://www.atetric.com/atetric/javadoc/com.hazelcast/hazelcast/3.5.4/src-html/com/hazelcast/core/IMap.html#line.75

源码片断：

public interface IMap<K, V> extends ConcurrentMap<K, V>, BaseMap<K, V> {
…
}

1. Distributed Map的工作原理

Hazelcast会将Map的元素进行间隔化，然后存放在各个节点上。每个节点会存放大约 = (1/n * total – data) +backups的个数，其中：n是cache server集群中节点（JVM）的数量。举例，如果我们有一个Map，有1000个元素，存放到由两个节点组成的server(group)上，那么每个节点会有500个元素，以及备份存上在另一个节点上的500个。

建立一个Hazelcast实例（一个节点），然后存放如下的元素：

public class FillMapMember {
  public static void main( String[] args ) { 
    HazelcastInstance hzInstance = Hazelcast.newHazelcastInstance();
    Map<String, String> capitalcities = hzInstance.getMap( "capitals" ); 
    capitalcities.put( "1", "Tokyo" );
    capitalcities.put( "2", "Paris” );
    capitalcities.put( "3", "Washington" );
    capitalcities.put( "4", "Ankara" );
    capitalcities.put( "5", "Brussels" );
    capitalcities.put( "6", "Amsterdam" );
    capitalcities.put( "7", "New Delhi" );
    capitalcities.put( "8", "London" );
    capitalcities.put( "9", "Berlin" );
    capitalcities.put( "10", "Oslo" );
    capitalcities.put( "11", "Moscow" );
    ...
    ...
    capitalcities.put( "120", "Stockholm" )
  }
}

运行以上代码，数据将被存入节点的间隔中：


注：有些间隔并不会有数据，因为我们只存放了120个元素，而Hazelcast默认分配的间隔总数有271个（这个数值可以通过修改环境变量hazelcast.partition.count来自定义）。

那么我们创建第二个节点，并重新运行以上代码。元素将被存入两个节点中，并相互进行备份。


HazelcastInstance::getMap返回了com.hazelcast.core.IMap的一个实例，在文章的一开始就提到了，IMap是继承了java.util.concurrent.ConruccrentMap接口。以下代码片断介绍了线程安全的方法ConcurrentMap.putIfAbsent(key, value)，ConcurrentMap.replace(key, value)的用法：

import com.hazelcast.core.Hazelcast;
import com.hazelcast.core.HazelcastInstance;
import java.util.concurrent.ConcurrentMap;

HazelcastInstance hazelcastInstance = Hazelcast.newHazelcastInstance();

Customer getCustomer( String id ) {
    ConcurrentMap<String, Customer> customers = hazelcastInstance.getMap( "customers" );
    Customer customer = customers.get( id );
    if (customer == null) {
        customer = new Customer( id );
        customer = customers.putIfAbsent( id, customer );
    }
    return customer;
}               

public boolean updateCustomer( Customer customer ) {
    ConcurrentMap<String, Customer> customers = hazelcastInstance.getMap( "customers" );
    return ( customers.replace( customer.getId(), customer ) != null );            
}

public boolean removeCustomer( Customer customer ) {
    ConcurrentMap<String, Customer> customers = hazelcastInstance.getMap( "customers" );
    return customers.remove( customer.getId(), customer );           
}

所有的ConcurrentMap的操作（例如put，remove）都要先将key进行lock，即只有一个线程在进行操作。但是无论怎样，都不会有死锁的情况，即最终都会返回success，ConcurrentMap的操作不会向外界抛出java.util.ConcurrentModificationException异常。

2. Map的备份

Distributed Map默认情况下会有一份备份。如果一个节点当掉了，数据不会丢会，这时候数据会立刻通过备份被恢复。所以当map.put(key, value)返回时，他是确保这个元素在集群的另一个节点是有备份的。对于读操作，也会确保map.get(key)返回的是最新的那个元素。

2.1 备份的一致性
为了提供数据的安全性，Hazelcast允许我们来设置备份的数量。也就是说，数据会从一台JVM拷贝到多台JVM上。以下是设置备份数量的代码：

<hazelcast>
  <map name="default">
    <backup-count>1</backup-count>
  </map>
</hazelcast>

注：备份的数量会增加内存的使用量。
注：备份的数据同步几乎是同时的。

2.2 从备份中读取数据
默认情况下，Hazelcast只有一个备份，如果备份数量超过1时，这时候每个节点会存放自己的数据以及其它节点上的备份。所以执行map.get(key)时，不止一个节点上有这个key，Hazelcast在默认情况下总是会去读真正存放元素的这个key，而不是备份数据。当参数read-back-data为true时，会从备份数据中读取。（默认情况下read-back-data参数值为false）。
注：当read-back-data为true时，可以提高效率）

<hazelcast>
  <map name="default">
    <backup-count>0</backup-count>
    <async-backup-count>1</async-backup-count>
    <read-backup-data>true</read-backup-data>
  </map>
</hazelcast>

3. Map的淘汰机制（eviction）

存放在Cache server的内存中的元素将会一直存在，除非手动的删除该Map中的元素，或是定义了缓存淘汰机制。Hazelcast支持对distributed map自定义淘汰机制，如：LRU（least recently used）和LFU（least frequently used）。

Map的淘汰机制（LRU, LFU）是建立在间隔的数量上的。比如我们定义了属性PER_NODE的 max-size，Hazelcast会自行计算能存放的元素的最大数量。缓存的淘汰机制就会以这个计算出来的数量为准。

3.1 Map淘汰机制的例子
假设我们有如下情况：

• 间隔(partition)数量为200
• 每个间隔存储元素(entry)的数量为100
• max-size(PER_NODE)设为：20000
• eviction-percentage为：10%

即我们现在已经存储的元素为20000（间隔数量 * 每个间隔的元素数量 = 200 * 100）。然后我们的最大存储max-size也是20000，这时候我们再往Server(节点group)中存放一个元素：

[a]元素存放到相应的间隔上。
间隔会检查实际存储的数量是否到达设定的最大值（max-size）。
[c]如果到达了（200*100+1 > 20000），那么大约100 * 10%=10个元素会被淘汰(evict)。

当我们再次检查map（拥有以上这些元素的）的数量，现在变成了~19990（约等于20000-10）。在这次淘汰之后，后面的put操作不会再次引起元素的淘汰，直到元素的数量（map的长度）又到达max-size。

注：以上的例子仅仅是为了说明缓存的淘汰机制是如何运作的，在实际情况中Hazelcast会根据群集中的节点数量以及具体的淘汰机制来确定需要被淘汰的最大（合适）数量。

[b]3.2 Map淘汰机制的配置
以下是淘汰机制配置的例子：

<hazelcast>
  <map name="default">
    ...
    <time-to-live-seconds>0</time-to-live-seconds>
    <max-idle-seconds>0</max-idle-seconds>
    <eviction-policy>LRU</eviction-policy>
    <max-size policy="PER_NODE">5000</max-size>
    <eviction-percentage>25</eviction-percentage>
    <min-eviction-check-millis>100</min-eviction-check-millis>
    ...
  </map>
</hazelcast>

解释：

• time-to-live: 单位是秒 (second)，默认为0秒（即不过期）。定义了元素存储的最大时间。如果不为0，那么存放时间超过定义的时间时就会被自动的淘汰。这个优先级比eviction-policy要高。
• max-idle-seceonds: 单位是秒 (second)，默认为0秒（即不过期）。定义了元素最大的闲暇时间。如果元素超过定义时间未被使用时，就会自动的被淘汰。
• eviction-policy: 淘汰机制如下:
        a. NONE: 默认配置，元素永远不会被淘汰，这时候max-size配置将会被忽略。但time-to-live-seconds和max-idle-seconds还是可以配置。
        b. LRU: Least Recently Used. 最久没有被使用的元素将会被淘汰。
        c. LFU: Least Frequently Used. 使用频率最少的元素将会被淘汰。
• max-size: 默认为0（即没有限制），定义了存储的最大数量。如果到达最大数量，那么将按eviction-policy配置的策略进行淘汰。如果希望该属性生效，那么eviction-policy不要设置为NONE。max-size有以下模式：
        a. PER_NODE: 默认模式。在每个JVM上最大能存储的数量。
                <max-size policy=”PER_NODE”>5000</max-size>
        b. PER_PARTITION: 每个间隔中最大能存储的数量。这个模式很少被使用。因为集群中的节点不一样的时候，每个JVM的间隔数量会不一样（集群中的节点越大，每个JVM的间隔数量越少），所以这个计算出来的值是会随着集群中的节点数量越多而减少。
                <max-size policy="PER_PARTITION">27100</max-size>
        c. USED_HEAP_SIZE: 每台JVM最大能使用的内存空间。
                <max-size policy="USED_HEAP_SIZE">4096</max-size>
        d. USED_HEAP_PERCENTAGE: 每台JVM最大能使用的内存空间百分比（占总的内存）。例如一台JVM被分配了1000M的内存，我们配置了10（%），那么能存储的元素总空间为100MB。
                <max-size policy="USED_HEAP_PERCENTAGE">10</max-size>
        e. FREE_HEAP_SIZE: 每台JVM的剩余有效空间。即剩余空间少于该配置值时就触发淘汰机制。
                <max-size policy="FREE_HEAP_SIZE">512</max-size>
        f. FREE_HEAP_PERCENTAGE: 每台JVM的剩余有效空间百分比（占JVM的总空间），计算方法同D
                <max-size policy="FREE_HEAP_PERCENTAGE">10</max-size>
• eviction-percentage: 值为0-100，默认为25。定义了当到达max-size时，即符合了触发淘汰机制的条件时，有多少百分比的元素将被淘汰。默认情况下，当需要淘汰元素时，将会有25%的元素被淘汰。
• min-eviction-check-millis: 单位是毫秒 (millisecond)，默认值为100。定义了多久检查一次间隔中的元素以使可以确定是否触发淘汰机制。当值设为0时，代表了每次put操作都会去检查。

总结起来就是多久确定一次元素在什么时候，有多少符合的目标元素会被淘汰。

注：当写操作比较频繁时，属性min-viction-check-millis应该要设的比较短一些（比写操作要的频率要短），这样就使得该被淘汰的元素在适当的时候就会被淘汰。

3.3 淘汰机制配置的例子
简单的例子，一个叫documents的map，如果其中一个JVM(节点)最大存储的数量超过了10000，那么就开始淘汰元素。淘汰的元素目标是使用次数最少的元素。同时如果一个元素在60秒内没有被使用，那么同样也会被淘汰。

<map name="documents">
  <max-size policy="PER_NODE">10000</max-size>
  <eviction-policy>LRU</eviction-policy> 
  <max-idle-seconds>60</max-idle-seconds>
</map>

3.4 淘汰指定的元素
3.3中的配置机制是针对map中所有的元素，那么如果需要淘汰指定的元素时，

myMap.put( "1", "John", 50, TimeUnit.SECONDS )

3.5 淘汰所有的元素
Hazelcast提供方法evictAll来删除map中所有的元素，除了被锁住的那些。如果定义了MapStore，那么删除所有元素不能用evictAll，而是clear()。
代码样例：

public class EvictAll {

    public static void main(String[] args) {
        final int numberOfKeysToLock = 4;
        final int numberOfEntriesToAdd = 1000;

        HazelcastInstance node1 = Hazelcast.newHazelcastInstance();
        HazelcastInstance node2 = Hazelcast.newHazelcastInstance();

        IMap<Integer, Integer> map = node1.getMap(EvictAll.class.getCanonicalName());
        for (int i = 0; i < numberOfEntriesToAdd; i++) {
            map.put(i, i);
        }

        for (int i = 0; i < numberOfKeysToLock; i++) {
            map.lock(i);
        }

        // should keep locked keys and evict all others.
        map.evictAll();

        System.out.printf("# After calling evictAll...\n");
        System.out.printf("# Expected map size\t: %d\n", numberOfKeysToLock);
        System.out.printf("# Actual map size\t: %d\n", map.size());

    }
}

4. 内存中的数据格式

IMap可以设置in-memory-format属性。默认情况下，Hazelcast存储的数据是二进制的（序列化对象）。但是有时候，这个存储格式会影响元素的值，（因为BINARY只是简单的进行序列化，然后存入Server，实例是同一个，所以我们在Client端改变元素的值，可能会影响Server上的值。）
通过属性in-memory-format，我们可以设置以下参数：

• BINARY (default)：默认配置。数据将被序列化后用二进制的形式存储。如果我们的数据经常进行读、写操作的话，可以使用这个配置。
• OBJECT：数据将会以对象的形式进行存储（deserizlized）。因为本身就是以对象的形式存储，那么在存储元素的操作中，不会进行

常规操作例如get需要返回对象的实例。当我们get一个存储格式为OBJECT的元素，这时候不会直接返回这个对象，而是需要进行克隆。该操作的步骤包括了：在元素存储的节点上进行序列化对象，然后在实际调用get操作的节点上进行反序列化对象。当我们get一个存储格式为BINARY的元素时，只需要在实际调用get操作的节点上进行反序列化即可，即就get操作来说，BINARY更快。

类似的，put操作在BINARY格式下更快（只需要序列化即可）。如果存储格式为OBJECT时，元素需要进行clone，即需要序列化与返序列化。
注：如果元素是以OBJECT存储的，那么改变返回的元素的值不会影响存储在Hazelcast server上的实例的值（因为有clone，实例不是同一个）。