import org.apache.curator.framework.CuratorFramework;

import org.apache.curator.framework.CuratorFrameworkFactory;

import org.apache.curator.framework.recipes.queue.*;

import org.apache.curator.framework.state.ConnectionState;

import org.apache.curator.retry.ExponentialBackoffRetry;

import java.io.UnsupportedEncodingException;

public class DistributedIdQueueDemo {

    private static CuratorFramework   client = CuratorFrameworkFactory.newClient("localhost:2181", new ExponentialBackoffRetry(3000, 2));

    private static String             path = "/queue/0002";

    private static DistributedIdQueue<String> queue;

    static {

        client.start();

    }

    public static void main(String[] args) throws Exception {

        QueueConsumer<String> consumer = new QueueConsumer<String>() {

            @Override

            public void stateChanged(CuratorFramework curatorFramework, ConnectionState connectionState) {

                System.out.println("state changed");

            }

            @Override

            public void consumeMessage(String s) {

                System.out.println("消费数据：" + s);

            }

        };

        queue = QueueBuilder.builder(client, consumer, new QueueSerializer<String>() {

            @Override

            public byte[] serialize(String s) {

                try {

                    return s.getBytes("utf-8");

                } catch (UnsupportedEncodingException e) {

                    e.printStackTrace();

                }

                return null;

            }

            @Override

            public String deserialize(byte[] bytes) {

                return new String(bytes);

            }

        }, path).buildIdQueue();

        queue.start();

        System.out.println("queue built");

        queue.put("a", "1");

        queue.put("b", "2");

        System.out.println("put ended");

        Thread.sleep(1000);

        queue.remove("1");

        queue.remove("2");

        System.out.println("remove ended");

        Thread.sleep(100);

        queue.close();

        Thread.sleep(50000);

        client.close();

    }

}