阿里云优惠
 汇集各种优惠信息 技术资料

首页    编程语言    【从入门到精通-ZooKeeper】ZooKeeper实战-分布式队列

【从入门到精通-ZooKeeper】ZooKeeper实战-分布式队列

创建时间:2019-09-16 10:08
浏览量:0
收藏

前言

上文【从入门到放弃-ZooKeeper】ZooKeeper入门中,我们学习了ZooKeeper的简单安装和cli使用。
接下来我们开始基于java API的实战编程。本文先来写一个分布式队列的代码实现。

设计

我们来写一个先进先出的分布式无界公平队列。参考我们之前介绍的【从入门到放弃-Java】并发编程-JUC-ConcurrentLinkedQueue和【从入门到放弃-Java】并发编程-JUC-LinkedBlockingQueue。我们直接继承AbstractQueue类,并实现Queue接口。
主要重写offer、poll、peek、size方法。
我们使用ZooKeeper的持久化顺序节点来实现分布式队列。
offer是入队,入队时新创建一个持久化顺序节点,节点后缀会根据ZooKeeper的特性自动累加。
poll的出队,获取根节点下的所有节点,根据后缀数字排序,数组最小的是最先入队的,因此要最先出队。
peek,获取到最下入队的数据,和poll的区别是,peek只获取数据,不出队,不删除已经消费的节点。
size获取队列长度,实现方式是,获取根节点下的节点数量即可。这个方法在并发时可能会有问题。慎用。

附: 申请阿里云服务器等产品时,可以使用2000元阿里云代金券

DistributedQueue

//继承AbstractQueue类并实现Queue接口
public class DistributedQueue<E> extends AbstractQueue<E> implements Queue<E> {
    private static Logger logger = LoggerFactory.getLogger(DistributedQueue.class);
    
    //ZooKeeper客户端,进行ZooKeeper操作
    private ZooKeeper zooKeeper;

    //根节点名称
    private String dir;

    //数据节点名称,顺序节点在插入口会变为 node{00000000xx} 格式
    private String node;

    //ZooKeeper鉴权信息
    private List<ACL> acls;

    /**
     * Constructor.
     *
     * @param zooKeeper the zoo keeper
     * @param dir       the dir
     * @param node      the node
     * @param acls      the acls
     */
    public DistributedQueue (ZooKeeper zooKeeper, String dir, String node, List<ACL> acls) {
        this.zooKeeper = zooKeeper;
        this.dir = dir;
        this.node = node;
        this.acls = acls;
        init();
    }


    private void init() {
        //需要先判断根节点是否存在,不存在的话,创建子节点时会出错。
        try {
            Stat stat = zooKeeper.exists(dir, false);
            if (stat == null) {
                zooKeeper.create(dir, null, acls, CreateMode.PERSISTENT);
            }
        } catch (Exception e) {
            logger.error("[DistributedQueue#init] error : " + e.toString(), e);
        }
    }
}

offer

/**
 * Offer boolean.
 *
 * @param o the o
 * @return the boolean
 */
@Override
public boolean offer(E o) {
    //构建要插入的节点名称
    String fullPath = dir.concat("/").concat(node);
    try {
        //创建子节点成功则返回入队成功
        zooKeeper.create(fullPath, objectToBytes(o), acls, CreateMode.PERSISTENT_SEQUENTIAL);
        return true;
    } catch (Exception e) {
        logger.error("[DistributedQueue#offer] error : " + e.toString(), e);
    }
    return false;
}

poll

/**
 * Poll e.
 *
 * @return the e
 */
@Override
public E poll() {
    try {
        //获取根节点所有子节点信息。
        List<String> children = zooKeeper.getChildren(dir, null);
        //如果队列是空的则返回null
        if (children == null || children.isEmpty()) {
            return null;
        }

        //将子节点名称排序
        Collections.sort(children);
        for (String child : children) {
            //拼接子节点的具体名称
            String fullPath = dir.concat("/").concat(child);
            try {
                //如果获取数据成功,则类型转换后,返回,并删除改队列中该节点
                byte[] bytes = zooKeeper.getData(fullPath, false, null);
                E data = (E) bytesToObject(bytes);
                zooKeeper.delete(fullPath, -1);
                return data;
            } catch (Exception e) {
                logger.warn("[DistributedQueue#poll] warn : " + e.toString(), e);
            }
        }

    } catch (Exception e) {
        logger.error("[DistributedQueue#peek] poll : " + e.toString(), e);
    }

    return null;
}

peek

/**
 * Peek e.
 *
 * @return the e
 */
@Override
public E peek() {
   
    try {
        //获取根节点所有子节点信息。
        List<String> children = zooKeeper.getChildren(dir, null);
        //如果队列是空的则返回null
        if (children == null || children.isEmpty()) {
            return null;
        }

        //将子节点名称排序
        Collections.sort(children);
        
        for (String child : children) {
            //拼接子节点的具体名称
            String fullPath = dir.concat("/").concat(child);
            try {
                //如果获取数据成功,则类型转换后,返回,不会删除改队列中该节点
                byte[] bytes = zooKeeper.getData(fullPath, false, null);
                E data = (E) bytesToObject(bytes);
                return data;
            } catch (Exception e) {
                logger.warn("[DistributedQueue#peek] warn : " + e.toString(), e);
            }
        }

    } catch (Exception e) {
        logger.error("[DistributedQueue#peek] warn : " + e.toString(), e);
    }

    return null;
}

size

/**
 * Size int.
 *
 * @return the int
 */
@Override
public int size() {
    try {
        //获取根节点的子节点名称
        List<String> children = zooKeeper.getChildren(dir, null);
        //返回子结点信息数量
        return children.size();
    } catch (Exception e) {
        logger.error("[DistributedQueue#offer] size : " + e.toString(), e);
    }

    return 0;
}

总结

上面我们一起学习了如何利用持久性顺序节点,创建一个分布式先进先出队列。

免费领取阿里云1888元代金券大礼包

 

阿里云新老用户均可领取!
自领取后:限时7天使用!

阿里云服务器2折优惠:低至293元/年

 

 

突发性能实例t5 1核1G:293元/年

突发性能实例t5 1核2G:459元/年

突发性能实例t5 2核4G:798元/年

共享型xn4实例1核1G内存:394元/年

共享型n4实例1核2G内存:653元/年

计算网络增强型实例2核4G内存:1566元/年

计算网络增强型实例4核8G内存:2991元/年

点此查看2折活动详情

阿里云高性能云服务器

 

 

网络增强型云服务器:2核4G ¥720元/年

高频应用云服务器:8核16G ¥4109元/年

本地SSD型云服务器:4核16G ¥6218.40元/年

大数据型云服务器:8核32G ¥11375.00元/年

GPU异构云服务器:16核40G ¥15563.00元/年

新用户满立减:每满1000立减50

 

1、到阿里云官网选购产品
2、加入到购物车
3、结算时立享满减

注意:新用户首次购买时必须先加到购物车,然后一起结算才享受此优惠。

腾讯云CVM云服务器22.07元起

 

 

腾讯云1核1G:22.07元/月、794.73元/3年

腾讯云2核2G:36.48元/月、1313.35元/3年

腾讯云2核4G:43.01元/月、1548.5元/3年

腾讯云4核8G:178.5元/月、6426元/3年