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 DistributedDelayQueueDemo {

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

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

    private static DistributedDelayQueue<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).buildDelayQueue();

        queue.start();

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

        // 当前时间 + 延迟时间 = 未来时间

        queue.put("a", System.currentTimeMillis() + 1000);

        queue.put("b", System.currentTimeMillis() + 20000);

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

        Thread.sleep(30000);

        queue.close();

        Thread.sleep(50000);

        client.close();

    }

}