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()
