2. 程式人生 > >跟著例項學習ZooKeeper的用法： 佇列

跟著例項學習ZooKeeper的用法： 佇列

阿新 • • 發佈：2018-12-22

使用Curator也可以簡化Ephemeral Node (臨時節點)的操作。Curator也提供ZK Recipe的分散式佇列實現。 利用ZK的 PERSISTENTSEQUENTIAL節點， 可以保證放入到佇列中的專案是按照順序排隊的。 如果單一的消費者從佇列中取資料， 那麼它是先入先出的，這也是佇列的特點。 如果你嚴格要求順序，你就的使用單一的消費者，可以使用leader選舉只讓leader作為唯一的消費者。

但是， 根據Netflix的Curator作者所說， ZooKeeper真心不適合做Queue，或者說ZK沒有實現一個好的Queue，詳細內容可以看 Tech Note 4， 原因有五：

  1. ZK有1MB 的傳輸限制。 實踐中ZNode必須相對較小，而佇列包含成千上萬的訊息，非常的大。
  2. 如果有很多節點，ZK啟動時相當的慢。 而使用queue會導致好多ZNode. 你需要顯著增大 initLimit 和 syncLimit.
  3. ZNode很大的時候很難清理。Netflix不得不建立了一個專門的程式做這事。
  4. 當很大量的包含成千上萬的子節點的ZNode時， ZK的效能變得不好
  5. ZK的資料庫完全放在記憶體中。 大量的Queue意味著會佔用很多的記憶體空間。

儘管如此， Curator還是建立了各種Queue的實現。 如果Queue的資料量不太多，資料量不太大的情況下，酌情考慮，還是可以使用的。

DistributedQueue

DistributedQueue是最普通的一種佇列。 它設計以下四個類：

  • QueueBuilder
  • QueueConsumer
  • QueueSerializer
  • DistributedQueue

建立佇列使用QueueBuilder,它也是其它佇列的建立類。

public static <T> QueueBuilder<T> builder(CuratorFramework client,
                                          QueueConsumer<T> consumer,
                                          QueueSerializer<T> serializer,
                                          java.lang.String queuePath)
 


QueueBuilder<MessageType>    builder = QueueBuilder.builder(client, consumer, serializer, path);
... more builder method calls as needed ...
DistributedQueue<MessageType queue = builder.build();    `

建立好queue就可以往裡面放入資料了：

queue.put(aMessage);

QueueSerializer提供了對佇列中的物件的序列化和反序列化。

QueueConsumer是消費者， 它可以接收佇列的資料。 處理佇列中的資料的程式碼邏輯可以放在QueueConsumer.consumeMessage()中。

正常情況下先將訊息從佇列中移除，再交給消費者消費。 但這是兩個步驟，不是原子的。 可以呼叫Builder的lockPath()消費者加鎖， 當消費者消費資料時持有鎖，這樣其它消費者不能消費此訊息。 如果消費失敗或者程序死掉，訊息可以交給其它程序。這會帶來一點效能的損失。 最好還是單消費者模式使用佇列。

例子：

package com.colobu.zkrecipe.queue;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.queue.DistributedQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
import org.apache.curator.utils.CloseableUtils;

public class DistributedQueueExample {
    private static final String PATH = "/example/queue";

    public static void main(String[] args) throws Exception {
        TestingServer server = new TestingServer();
        CuratorFramework client = null;
        DistributedQueue<String> queue = null;
        try {
            client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.getCuratorListenable().addListener(new CuratorListener() {
                @Override
                public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
                    System.out.println("CuratorEvent: " + event.getType().name());
                }
            });

            client.start();
            QueueConsumer<String> consumer = createQueueConsumer();
            QueueBuilder<String> builder = QueueBuilder.builder(client, consumer, createQueueSerializer(), PATH);
            queue = builder.buildQueue();
            queue.start();

            for (int i = 0; i < 10; i++) {
                queue.put(" test-" + i);
                Thread.sleep((long)(3 * Math.random()));
            }

            Thread.sleep(20000);

        } catch (Exception ex) {

        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
            CloseableUtils.closeQuietly(server);
        }
    }

    private static QueueSerializer<String> createQueueSerializer() {
        return new QueueSerializer<String>(){

            @Override
            public byte[] serialize(String item) {
                return item.getBytes();
            }

            @Override
            public String deserialize(byte[] bytes) {
                return new String(bytes);
            }

        };
    }

    private static QueueConsumer<String> createQueueConsumer() {

        return new QueueConsumer<String>(){

            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.out.println("connection new state: " + newState.name());
            }

            @Override
            public void consumeMessage(String message) throws Exception {
                System.out.println("consume one message: " + message);                
            }

        };
    }
}

DistributedIdQueue

DistributedIdQueue和上面的佇列類似， 但是可以為佇列中的每一個元素設定一個ID。 可以通過ID把佇列中任意的元素移除。 它涉及幾個類：

  • QueueBuilder
  • QueueConsumer
  • QueueSerializer
  • DistributedQueue

通過下面方法建立：

builder.buildIdQueue()

放入元素時：

queue.put(aMessage, messageId);

移除元素時：

int numberRemoved = queue.remove(messageId);

在這個例子中， 有些元素還沒有被消費者消費時就移除了，這樣消費者不會收到刪除的訊息。

package com.colobu.zkrecipe.queue;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.queue.DistributedIdQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
import org.apache.curator.utils.CloseableUtils;

public class DistributedIdQueueExample {
    private static final String PATH = "/example/queue";

    public static void main(String[] args) throws Exception {
        TestingServer server = new TestingServer();
        CuratorFramework client = null;
        DistributedIdQueue<String> queue = null;
        try {
            client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.getCuratorListenable().addListener(new CuratorListener() {
                @Override
                public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
                    System.out.println("CuratorEvent: " + event.getType().name());
                }
            });

            client.start();
            QueueConsumer<String> consumer = createQueueConsumer();
            QueueBuilder<String> builder = QueueBuilder.builder(client, consumer, createQueueSerializer(), PATH);
            queue = builder.buildIdQueue();
            queue.start();

            for (int i = 0; i < 10; i++) {
                queue.put(" test-" + i, "Id" + i);
                Thread.sleep((long)(50 * Math.random()));
                queue.remove("Id" + i);
            }

            Thread.sleep(20000);

        } catch (Exception ex) {

        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
            CloseableUtils.closeQuietly(server);
        }
    }

    private static QueueSerializer<String> createQueueSerializer() {
        return new QueueSerializer<String>(){

            @Override
            public byte[] serialize(String item) {
                return item.getBytes();
            }

            @Override
            public String deserialize(byte[] bytes) {
                return new String(bytes);
            }

        };
    }

    private static QueueConsumer<String> createQueueConsumer() {

        return new QueueConsumer<String>(){

            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.out.println("connection new state: " + newState.name());
            }

            @Override
            public void consumeMessage(String message) throws Exception {
                System.out.println("consume one message: " + message);                
            }

        };
    }
}

DistributedPriorityQueue

優先順序佇列對佇列中的元素按照優先順序進行排序。 Priority越小， 元素月靠前， 越先被消費掉。 它涉及下面幾個類：

  • QueueBuilder
  • QueueConsumer
  • QueueSerializer
  • DistributedPriorityQueue

通過builder.buildPriorityQueue(minItemsBeforeRefresh)方法建立。 當優先順序佇列得到元素增刪訊息時，它會暫停處理當前的元素佇列，然後重新整理佇列。minItemsBeforeRefresh指定重新整理前當前活動的佇列的最小數量。 主要設定你的程式可以容忍的不排序的最小值。

放入佇列時需要指定優先順序：

queue.put(aMessage, priority);

例子：

package com.colobu.zkrecipe.queue;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.queue.DistributedPriorityQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
import org.apache.curator.utils.CloseableUtils;

public class DistributedPriorityQueueExample {
    private static final String PATH = "/example/queue";

    public static void main(String[] args) throws Exception {
        TestingServer server = new TestingServer();
        CuratorFramework client = null;
        DistributedPriorityQueue<String> queue = null;
        try {
            client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.getCuratorListenable().addListener(new CuratorListener() {
                @Override
                public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
                    System.out.println("CuratorEvent: " + event.getType().name());
                }
            });

            client.start();
            QueueConsumer<String> consumer = createQueueConsumer();
            QueueBuilder<String> builder = QueueBuilder.builder(client, consumer, createQueueSerializer(), PATH);
            queue = builder.buildPriorityQueue(0);
            queue.start();

            for (int i = 0; i < 10; i++) {
                int priority = (int)(Math.random() * 100);
                System.out.println("test-" + i + " priority:" + priority);
                queue.put("test-" + i, priority);
                Thread.sleep((long)(50 * Math.random()));
            }

            Thread.sleep(20000);

        } catch (Exception ex) {

        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
            CloseableUtils.closeQuietly(server);
        }
    }

    private static QueueSerializer<String> createQueueSerializer() {
        return new QueueSerializer<String>(){

            @Override
            public byte[] serialize(String item) {
                return item.getBytes();
            }

            @Override
            public String deserialize(byte[] bytes) {
                return new String(bytes);
            }

        };
    }

    private static QueueConsumer<String> createQueueConsumer() {

        return new QueueConsumer<String>(){

            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.out.println("connection new state: " + newState.name());
            }

            @Override
            public void consumeMessage(String message) throws Exception {
                System.out.println("consume one message: " + message);                
            }

        };
    }
}

DistributedDelayQueue

JDK中也有DelayQueue，不知道你是否熟悉。 DistributedDelayQueue也提供了類似的功能， 元素有個delay值， 消費者隔一段時間才能收到元素。 涉及到下面四個類。

  • QueueBuilder
  • QueueConsumer
  • QueueSerializer
  • DistributedDelayQueue

通過下面的語句建立：

QueueBuilder<MessageType>    builder = QueueBuilder.builder(client, consumer, serializer, path);
... more builder method calls as needed ...
DistributedDelayQueue<MessageType> queue = builder.buildDelayQueue();

放入元素時可以指定delayUntilEpoch：

queue.put(aMessage, delayUntilEpoch);

注意delayUntilEpoch不是離現在的一個時間間隔， 比如20毫秒，而是未來的一個時間戳，如 System.currentTimeMillis() + 10秒。 如果delayUntilEpoch的時間已經過去，訊息會立刻被消費者接收。

例子：

package com.colobu.zkrecipe.queue;

import java.util.Date;

import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.CuratorEvent;
import org.apache.curator.framework.api.CuratorListener;
import org.apache.curator.framework.recipes.queue.DistributedDelayQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.test.TestingServer;
import org.apache.curator.utils.CloseableUtils;

public class DistributedDelayQueueExample {
    private static final String PATH = "/example/queue";

    public static void main(String[] args) throws Exception {
        TestingServer server = new TestingServer();
        CuratorFramework client = null;
        DistributedDelayQueue<String> queue = null;
        try {
            client = CuratorFrameworkFactory.newClient(server.getConnectString(), new ExponentialBackoffRetry(1000, 3));
            client.getCuratorListenable().addListener(new CuratorListener() {
                @Override
                public void eventReceived(CuratorFramework client, CuratorEvent event) throws Exception {
                    System.out.println("CuratorEvent: " + event.getType().name());
                }
            });

            client.start();
            QueueConsumer<String> consumer = createQueueConsumer();
            QueueBuilder<String> builder = QueueBuilder.builder(client, consumer, createQueueSerializer(), PATH);
            queue = builder.buildDelayQueue();
            queue.start();

            for (int i = 0; i < 10; i++) {
                queue.put("test-" + i, System.currentTimeMillis() + 10000);
            }
            System.out.println(new Date().getTime() + ": already put all items");


            Thread.sleep(20000);

        } catch (Exception ex) {

        } finally {
            CloseableUtils.closeQuietly(queue);
            CloseableUtils.closeQuietly(client);
            CloseableUtils.closeQuietly(server);
        }
    }

    private static QueueSerializer<String> createQueueSerializer() {
        return new QueueSerializer<String>() {

            @Override
            public byte[] serialize(String item) {
                return item.getBytes();
            }

            @Override
            public String deserialize(byte[] bytes) {
                return new String(bytes);
            }

        };
    }

    private static QueueConsumer<String> createQueueConsumer() {

        return new QueueConsumer<String>() {

            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                System.out.println("connection new state: " + newState.name());
            }

            @Override
            public void consumeMessage(String message) throws Exception {
                System.out.println(new Date().getTime() + ": consume one message: " + message);
            }

        };
    }
}

SimpleDistributedQueue

前面雖然實現了各種佇列，但是你注意到沒有，這些佇列並沒有實現類似JDK一樣的介面。 SimpleDistributedQueue提供了和JDK一致性的介面(但是沒有實現Queue介面)。 建立很簡單：

public SimpleDistributedQueue(CuratorFramework client,
                              String path)

增加元素：

public boolean offer(byte[] data) throws Exception

刪除元素：

public byte[] take() throws Exception

另外還提供了其它方法：

public byte[] peek() throws Exception
public byte[] poll(long timeout, TimeUnit unit) throws Exception
public byte[] poll() throws Exception
public byte[] remove() throws Exception
public byte[] element() throws Exception

沒有add方法， 多了take方法。

take方法在成功返回之前會被阻塞。 而poll在佇列為空時直接返回null。w String(client.getData().forPath(PATH2)));

    } catch (Exception ex) {
        ex.printStackTrace();
    } finally {
        CloseableUtils.closeQuietly(node);
        CloseableUtils.closeQuietly(client);
        CloseableUtils.closeQuietly(server);
    }

}

} “`


相關推薦

跟著例項學習ZooKeeper的用法： 佇列

使用Curator也可以簡化Ephemeral Node (臨時節點)的操作。Curator也提供ZK Recipe的分散式佇列實現。 利用ZK的 PERSISTENTSEQUENTIAL節點， 可以保證放入到佇列中的專案是按照順序排隊的。 如果單一的消費者從佇列中取資料， 那麼它是先入先出的

跟著例項學習ZooKeeper的用法： Leader選舉

ZooKeeper官方給出了使用zookeeper的幾種用途。 Leader Election Barriers Queues Locks Two-phased Commit 其它應用如Name Service, Configuration, Group Membership 在實際使用

跟著例項學習ZooKeeper的用法： Curator擴充套件庫

還記得Curator提供哪幾個元件嗎？ 我們不妨回顧一下： Recipes Framework Utilities Client Errors Extensions 前面的例子其實前五個元件都涉及到了， 比如Utilities例子的TestServer, Client裡的CuratorZo

跟著例項學習ZooKeeper的用法： 計數器

這一篇文章我們將學習使用Curator來實現計數器。 顧名思義，計數器是用來計數的, 利用ZooKeeper可以實現一個叢集共享的計數器。 只要使用相同的path就可以得到最新的計數器值， 這是由ZooKeeper的一致性保證的。Curator有兩個計數器， 一個是用int來計數，一個用lon

跟著例項學習ZooKeeper的用法： Barrier

分散式Barrier是這樣一個類： 它會阻塞所有節點上的等待程序，知道某一個被滿足， 然後所有的節點繼續進行。 比如賽馬比賽中， 等賽馬陸續來到起跑線前。 一聲令下，所有的賽馬都飛奔而出。 柵欄Barrier DistributedBarrier類實現了柵欄的功能。 它的建構函式如下： pu

跟著例項學習ZooKeeper的用法： 臨時節點

使用Curator也可以簡化Ephemeral Node (臨時節點)的操作。 臨時節點駐存在ZooKeeper中，當連線和session斷掉時被刪除。 比如通過ZooKeeper釋出服務，服務啟動時將自己的資訊註冊為臨時節點，當服務斷掉時ZooKeeper將此臨時節點刪除，這樣client就

跟著例項學習ZooKeeper的用法： 快取

可以利用ZooKeeper在叢集的各個節點之間快取資料。 每個節點都可以得到最新的快取的資料。 Curator提供了三種類型的快取方式：Path Cache,Node Cache 和Tree Cache。 Path Cache Path Cache用來監控一個ZNode的子節點. 當一個子節點

跟著例項學習ZooKeeper的用法： 分散式鎖

鎖 分散式的鎖全域性同步， 這意味著任何一個時間點不會有兩個客戶端都擁有相同的鎖。 可重入鎖Shared Reentrant Lock 首先我們先看一個全域性可重入的鎖。 Shared意味著鎖是全域性可見的， 客戶端都可以請求鎖。 Reentrant和JDK的ReentrantLock類似，

跟著例項學習ZooKeeper的用法： Curator框架應用

前面的幾篇文章介紹了一些ZooKeeper的應用方法， 本文將介紹Curator訪問ZooKeeper的一些基本方法， 而不僅僅限於指定的Recipes， 你可以使用Curator API任意的訪問ZooKeeper。 CuratorFramework Curator框架提供了一套高階的API

帶你用例項學習代理模式：靜態代理、動態代理（JDK、CGlib）以及區別和優缺點

Spring AOP的核心技術就是動態代理，所以小編學習並整理了代理模式的材料，供大家一起學習。 1、代理模式滿足的三個必要條件： 兩個角色：執行者、被代理物件 這個過程必須要做，但是自己不能做或者不想做，交給專業的人（媒婆） 執行者必須拿到被代理物件的引用（需要知道你要什

演算法學習之五：佇列

佇列的概念: 佇列是一種特殊的線性表，特殊之處在於它只允許在表的前端（front）進行刪除操作，而在表的後端（rear）進行插入操作。進行插入操作的端稱為隊尾，進行刪除操作的端稱為隊頭。佇列中沒有元素時，稱為空佇列。 要著重強調的地方： （1）所有刪除的操

跟著例項學習java多執行緒3-synchronized的多種寫法有何區別？

同步程式碼塊是一種有效實現操作原子性的方法，上一章我們講了一些同步的原子操作的基礎。 現在我們回憶一下上一章的兩個問題。 1：不同的synchronized的寫法有什麼區別，又該怎麼寫建立執行緒的程式碼呢？ 以class例項物件作為鎖的寫法 寫法1 package com.

分散式學習筆記七：基於zookeeper實現分散式鎖

一、分散式鎖介紹         分散式鎖主要用於在分散式環境中保護跨程序、跨主機、跨網路的共享資源實現互斥訪問，以達到保證資料的一致性。 二、架構介紹     &nb

分散式學習筆記六：分散式利器Zookeeper

ZooKeeper簡介 ZooKeeper是一個開放原始碼的分散式應用程式協調服務，它包含一個簡單的原語集，分散式應用程式可以基於它實現同步服務，配置維護和命名服務等。   ZooKeeper設計目的 1.最終一致性：client不論連線到哪個Server，展示給它都

python 學習彙總60：類例項作為函式的引數（初級學習- tcy）

 類例項作為函式的引數 2018/11/20  使用輸出引數（通過引用呼叫）編寫函式 ： class callByRef: def __init__(self, **args): for (key, value) in args.items(): setatt

Java入門系列：例項講解ArrayList用法

1.ArrayList建構函式 ArrayList類支援3個構造方法。 Arraylist() 這個構造方法構造了一個空的連結串列。 ArrayList(Collection<? extends E> c) 這個構造方法構造了一個包含指定元素集合的連結串列，

zookeeper學習心得一：基礎講解---zk節點型別以及對節點的相應操作

zookeeper節點型別     持久節點（PERSISTENT）     所謂持久節點，是指在節點建立後，就一直存在，直到有刪除操作來主動清除這個節點——不會因為建立該節點的客戶端會話失效而消失。    

JUnit學習筆記(一)：基本用法

JUnit學習筆記(一)：基本用法 JUnit是Java的單元測試框架。JUnit提倡“先測試後編碼”的理念，強調建立測試資料的程式碼，可以先測試，再應用。 框架的特點：     JUnit

JBPM4.4學習之二：第一個流程例項

1)建立專案和相關配置 在eclipse中建立一個Java Project，將jbpm4.jar和lib資料夾下的第三方jar包Add到專案中。 按照個人習慣，建立config和jbpm兩個原始檔夾，用於放配置檔案和jbpm流程定義檔案。 將jbpm-4.4/exam

Java學習筆記19：==與equals用法比較

String s1="abc"; String s2="abc"; System.out.println(s1==s2);//true System.out.println(s1.eqauls(s2));//true //建立了幾個物件？ String s1=new Strin