[Hazelcast系列五] 分布式queue

Hazelcast queue是java.util.concurrent.BlockingQueue 的分布式实现，Hazelcast的queue允许集群所有成员和其交互。使用Hazelcast的分布式queue，可以一个成员添加元素，另外的成员移除元素。

7.3.1. 获取一个队列并向队列添加元素

使用Hazelcast实例的getQueue 获得一个queue，并调用queue的put 方法向队列添加元素。

HazelcastInstance instance = Hazelcast.newHazelcastInstance();
IQueue<Integer> queue = instance.getQueue("data");
queue.put(1);
Integer item = queue.take();
System.out.println(item);
boolean offerResult = queue.offer(2,1, TimeUnit.SECONDS);
System.out.println(offerResult);
Integer nextItem = queue.poll(1,TimeUnit.SECONDS);
System.out.println(nextItem);

集群内所有的操作都满足FIFO的顺序，用户自定义的对象出队和入队都需要序列化。Hazelcast分布式的queue不支持批量变量队列元素。所有的元素都会拷贝到实例本地，实例对queue的遍历都是本地遍历。可以使用ItemListener ` 监听queue中元素的添加和移除。

7.3.2. 创建样例队列

下面的样例代码展示了使用Hazelcast的分布式queue连接消费者和生产者。

生产者生产数据

生产者每隔1s向队列中添加100个整数，添加完成后向队列插入-1表示结束：

public class Producer {
    public static void main(String[] args) throws Exception {
        HazelcastInstance instance = Hazelcast.newHazelcastInstance();
        IQueue<Integer> queue = instance.getQueue("data");
        for (int i = 0; i < 100; i++) {
            queue.put(i);
            TimeUnit.SECONDS.sleep(1);
        }
        queue.put(-1);
    }
}

消费者消费数据

消费者每3s从queue中读取一个数据：

public class FirstConsumer {
    public static void main(String[] args) throws Exception {
        HazelcastInstance instance = Hazelcast.newHazelcastInstance();
        IQueue<Integer> queue = instance.getQueue("data");
        while (true) {
            Integer val = queue.poll();
            System.out.println(val);
            assert val != null;
            if (val.equals(-1)) {
                System.out.println("consume finished");
                break;
            }
            TimeUnit.SECONDS.sleep(3);
        }
    }
}

平衡生产者和消费者

上面的生产者每1s生产一条数据，消费者每3s消费一条数据，因此队列中的数据会一直增加。为了平衡生产和消费速率的不匹配，需要启动多个消费者来增加消费速度。

启动两个消费者，个消费者的输出如下：

...
0
2
4
...

第二个消费的输出如下：

...
1
3
5
...

在一个queue有多个生产者和消费的场景，可以使用多个queue解决queue瓶颈问题。使用多queue的场景，发送到多个queue数据的顺序无法保证。因为大多数场景严格的有序并不是那么重要，因此多queue是一个不错的解决方案。

从queue读取数据的顺序和数据添加到队列的顺序一致，但是有多个消费者时，这个顺序无法得到保证

ItemIDs

向queue添加item时，Hazelcast会给每个item分配一个有序递增的itemId 。可以从下面几点理解itemId 的分配行为：

一个Hazelcast节点有一个queue，而且queue的备份数至少为1，节点发生重启时，Hazelcast会从重启前已知的大itemId 开始分配，不会从小值开始分配新的itemId 。
如果整个集群-发生了重启，而且queue设置了持久化数据存储QueueStore ，会使用上述的itemId 分配策略。Hazelcast会使用loadAllKeys 返回的大ID开始分配，如果loadAllKeys 没有任何返回值，集群从小值开始分配。
itemId 保证在内存和持久存储中无重复。

7.3.3. 设置有界队列

有界队列是一个容量受限的队列，当有界队列满时，在元素被移除队列之前任何元素都不能添加到队列中。通过max-size 设置容量即可将Hazelcast中的一个分布式队列变为有界队列。max-size 定义了queue可以存储的大数据量。一旦queue中的数据量达到该值，put 操作将会被阻塞直到queue的数据量低于max-size。

下面是一个容量为10的queue配置：

<hazelcast>
    ...
    <queue name="queue">
        <max-size>10</max-size>
    </queue>
    ...
</hazelcast>

7.3.4. 持久存储queue数据

Hazelcast通过QueueStore 可以从持久存储中加载数据到queue/queue数据写入持久存储。如果设置了queue存储，向queue添加的数据也会被添加到持久存储中。如果queue中的元素超过了内存限制，接下来添加的元素会存储在持久存储中而不存储在内存中。

QueueStore 提供了store 、storeAll 、load 和delete方法用于存储、加载和删除queue存储在持久存储中的数据。下面是一个简单的QueueStore 实现：

public class SimpleQueueStore implements QueueStore<Item> {

    @Override
    public void delete(Long key) {
        System.out.println("delete");
    }

    @Override
    public void store(Long key, Item value) {
        System.out.println("store");
    }

    @Override
    public void storeAll(Map<Long, Item> map) {
        System.out.println("store all");
    }

    @Override
    public void deleteAll(Collection<Long> keys) {
        System.out.println("deleteAll");
    }

    @Override
    public Item load(Long key) {
        System.out.println("load");
        return null;
    }

    @Override
    public Map<Long, Item> loadAll(Collection<Long> keys) {
        System.out.println("loadALl");
        return null;
    }

    @Override
    public Set<Long> loadAllKeys() {
        System.out.println("loadAllKeys");
        return null;
    }
}

Item 必须支持序列化. 下面是一个queue支持持久存储数据的配置：

<hazelcast>
    ...
    <queue name="queue">
        <max-size>10</max-size>
        <queue-store>
            <class-name>com.hazelcast.QueueStoreImpl</class-name>
            <properties>
                <property name="binary">false</property>
                <property name="memory-limit">1000</property>
                <property name="bulk-load">500</property>
            </properties>
        </queue-store>
    </queue>
    ...
</hazelcast>

下面是配置参数的简单解释：

Binary: 存储格式
Memory Limit: 存储在queue内存和持久存储中的item数量，超过数量的item将只存储在持久存储中。
Bulk Load: 批加载大小。

7.3.5. 脑裂保护

可以配置在应用queue操作前检查集群小可用节点数，检查可以避免在网络分区产生的场景下queue的操作可以在所有集群都操作成功。

下面的操作支持脑裂保护检查：

WRITE, READ_WRITE
- Collection.addAll()
- Collection.removeAll(), Collection.retainAll()
- BlockingQueue.offer(), BlockingQueue.add(), BlockingQueue.put()
- BlockingQueue.drainTo()
- IQueue.poll(), Queue.remove(), IQueue.take()
- BlockingQueue.remove()
READ, READ_WRITE
- Collection.clear()
- Collection.containsAll(), BlockingQueue.contains()
- Collection.isEmpty()
- Collection.iterator(), Collection.toArray()
- Queue.peek(), Queue.element()
- Collection.size()
- BlockingQueue.remainingCapacity()