七curator recipes之阻塞队列SimpleDistributedQueue

Posted 2021-02-03 lay2017

简介

Java在单机环境实现了BlockQueue阻塞队列，与之类似的curator实现了分布式场景下的阻塞队列，SimpleDistributedQueue

官方文档：http://curator.apache.org/curator-recipes/simple-distributed-queue.html

javaDoc：http://curator.apache.org/apidocs/org/apache/curator/framework/recipes/queue/SimpleDistributedQueue.html

注意：zookeeper虽然可以实现队列，但是官方并不推荐使用zookeeper来做队列，具体请参考官网

代码示例

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.queue.SimpleDistributedQueue;
import org.apache.curator.retry.ExponentialBackoffRetry;

public class SimpleQueueDemo {
    private static CuratorFramework client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(3000, 2));
    private static String path = "/queue/path001";
    public static void main(String[] args) throws InterruptedException {
        client.start();
        System.out.println("started");
        SimpleDistributedQueue queue = new SimpleDistributedQueue(client, path);
        new Thread(() -> {
            try {
                System.out.println("sleeping");
                Thread.sleep(3000);
                System.out.println("sleep end");
                new SimpleDistributedQueue(client, path).offer("lay".getBytes("utf-8"));
                System.out.println("offered");
            } catch (Exception e) {
                System.out.println("exception");
                e.printStackTrace();
            }
        }).start();
        System.out.println("polling");
        String data = null;
        try {
            data = new String(queue.take());
        } catch (Exception e) {
            e.printStackTrace();
        }
        System.out.println("data=" + data);
        client.close();
    }
}

输出结果

started
polling
sleeping
sleep end
offered
data=lay

主线程会阻塞直到offer了数据