[Hazelcast系列 三]分布式数据结构简介

Hazelcast提供了很多通用数据结构的分布式实现。针对每一种不同的客户端语言，Hazelcast都尽量模拟该语言原生接口，对于Java客户端，Hazelcast提供的IMap 和java.util.Map 有近似的语义。为了更加直观的描述，针对Hazelcast中的数据结构，我们将提供Java中等效或类似接口。 所有这些结构都可以在Java，.NET，C ++，Node.js，Python，Go和Scala客户端中使用。

• 标准集合

  • Map： java.util.Map 的分布式实现。
  • Queue： java.util.concurrent.BlockingQueue 的分布式实现。
  • Ringbuffer：Java中没有对应的数据结构，Ringbuffer通常用于可靠的事件系统。
  • Set： java.util.Set 的分布式并发实现。
  • List： java.util.List 的分布式并发实现，支持存储重复元素，这是和Set的唯一区别。
  • Multimap： com.google.common.collect.Multimap 的分布式实现，支持存储重复键。
  • Replicated Map：不支持分区的Map数据结构，集群所有成员都有全量数据。
  • Cardinality Estimator ：实现了lajolet’s HyperLogLog 算法的数据结构。

• 主题

  主题是用于多个订阅者的分布式消息分发机制，和kakfa、pulsar中的主题一样，同样也是pub/sub消息模型的关键。Hazelcast支持通过主题进行消息的可靠分发。

• 并发工具

  • FencedLock：java 中 java.util.concurrent.locks.Lock的分布式实现，可以保证集群中只有一个线程可以获得锁。
  • ISemaphore：java中 java.util.concurrent.Semaphore的分布式实现。
  • IAtomicLong ：java中 java.util.concurrent.atomic.AtomicLong的分布式实现。
  • IAtomicReference：Java中java.util.concurrent.atomic.AtomicReference的分布式实现。
  • FlakeIdGenerator：用于生产集群范围内的唯一标识符。
  • ICountdownLatch： Java中 java.util.concurrent.CountDownLatch的分布式实现。
  • PN counter：一个分布式数据结构，其中每个Hazelcast实例都可以递增和递减计数器值，并将这些更新传播到所有副本。

• Event Journal：是一种分布式数据结构，用于存储map或缓存上操作的历史记录。

1. 分布式对象简介

基于分区策略，Hazelcast有两种类型的分布式数据结构：分区数据结构和非分区数据结构，分区数据结构一个分区只保存部分数据，非分区数据结构一个分区保存全部的数据。

Hazelcast中的分区数据结构包括以下四种：

• Map
• MultiMap
• Cache
• Event Journal

以下为非分区数据结构：

• Queue
• Set
• List
• Ringbuffer
• FencedLock
• ISemaphore
• IAtomicLong
• IAtomicReference
• FlakeldGenerator
• ICountdownLatch
• Cardinality Estimator
• PN Counter

除分区和非分区数据结构以外，Hazelcast同时提供了Replicated Map（可复制集合）数据结构。

2 . 分布式对象的加载和销毁

针对大多数分布式对象，Hazelcast提供了获取实例的get 方法。一个分布式对象的加载，首先需要创建一个Hazelcast实例，然后通过Hazelcast实例调用对应的get 方法获取。下面的例子创建一个Hazelcast实例instance1 并在该实例上创建一个叫data的分布式Map。

HazelcastInstance instance1 = Hazelcast.newHazelcastInstance();
IMap<String, String> data = instance1.getMap("data");

Hazelcast支持对分布式对象的属性进行配置，默认使用hazelcast.xml中的配置信息，也可以在xml中显示配置或根据需求使用代码进行配置。Hazelcast中的大多数分布式对象都是懒加载，只有在第一次操作对象时才创建对象。使用已经创建的对象，可以直接通过对象的引用访问，无需重新加载一次。

销毁分布式对象可以使用destory 方法，该方法会清理和释放对象的所有资源。使用该方法时需要十分谨慎，销毁对象并创建一个新的对象赋值给原来的对象引用不会产生任何错误。下面的代码展示了这种潜在的危险。

HazelcastInstance instance1 = Hazelcast.newHazelcastInstance();
IMap<String, String> data = instance1.getMap("data");
data.put("hazelcast","a good tool");
System.out.println("The size of map = " + data.size());
data.destroy();
System.out.println("The size of map = " + data.size());

上述代码运行没有任何错误，输出结果如下：

The size of map = 1
The size of map = 0

Hazelcast中所有的分布式数据结构都被设计为当访问对象的时刻才创建对象，因此即便已经销毁了对象，只要对该对象还有访问，对象就会被重新创建。

3. 控制分区

Hazelcast使用分布式对象的名字决定该对象应该存储在哪个分区。下面创建两个Queue 对象，名字分别为q1和q2：

HazelcastInstance instance1 = Hazelcast.newHazelcastInstance();
IQueue<String> queue1 = instance1.getQueue("q1");
IQueue<String> queue2 = instance1.getQueue("q2");

由于queue1和queue2两个队列的名字不同，因此他们将会存储在不同的分区。如果想将两个队列存储在同一个分区，可以使用@ 符设置分区key，从而将两个队列存储在同一个分区：

HazelcastInstance instance1 = Hazelcast.newHazelcastInstance();
IQueue<String> queue1 = instance1.getQueue("q1@test");
IQueue<String> queue2 = instance1.getQueue("q2@test");

现在两个队列存储在同一个分区，因为它们使用相同的分区key ：test，Hazelcast提供了getPartitionKey方法用于获取分区key的信息，这将有助于创建一个和已有对象存储在同一个分区的新对象。

HazelcastInstance instance1 = Hazelcast.newHazelcastInstance();
IQueue<String> queue1 = instance1.getQueue("q1@test");
IQueue<String> queue2 = instance1.getQueue("q2@test");
System.out.println("queue1 partition key = " + queue1.getPartitionKey());
System.out.println("queue2 partition key = " + queue2.getPartitionKey());

上面的代码将会输出queue1和queue2的分区key：

queue1 partition key = test
queue2 partition key = test

4. 公共特性

Hazelcast中所有分布式对象都具有高可用性（HA）：

• 当集群中有一个成员发生故障时，保存相同数据的备份副本会将包括权限、锁在内的所有数据分配给集群内其他成员，无数据丢失。
• 集群中所有的成员都有相同的权力和责任，没有主节点或leader，不依赖外部服务，因此没有单点故障。

5. 样例

下面的代码将简单的展示如何获取已经创建的分布式对象实例以及如何监听实例事件：

HazelcastInstance instance = Hazelcast.newHazelcastInstance();
instance.addDistributedObjectListener(new DistributedObjectListener() {
    @Override
    public void distributedObjectCreated(DistributedObjectEvent event) {
        DistributedObject instance = event.getDistributedObject();
        System.out.println(instance.getName() + " created");
    }

    @Override
    public void distributedObjectDestroyed(DistributedObjectEvent event) {
        System.out.println(event.getObjectName() + " destroyed");
    }
});
IQueue<String> queue1 = instance.getQueue("q1");
IQueue<String> queue2 = instance.getQueue("q2");
queue1.add("hello");
queue2.add("world");
instance.getDistributedObjects().forEach(o -> System.out.println(o.getName()));
instance.getDistributedObjects().forEach(DistributedObject::destroy);

上面的样例代码将输出：

q1 created
q2 created

q1
q2

q1 destroyed
q2 destroyed