Apache Zookeeper. Паттерны использования.

class Zookeeper {
    String create(String path, byte data[], List<ACL> acl, CreateMode createMode);
    Stat setData(String path, byte data[], int version);
    void delete(String path, int version);

    Stat exists(String path, Watcher watcher);
    List<String> getChildren(String path, Watcher watcher, Stat stat);
    byte[] getData(String path, Watcher watcher, Stat stat);
}

class WatchedEvent {
    KeeperState keeperState;
    EventType eventType;
    String path;
}

interface Watcher {
    void process(WatchedEvent event);
    interface Event {
        static enum EventType {
            None(-1),
            NodeCreated(1),
            NodeDeleted(2),
            NodeDataChanged(3),
            NodeChildrenChanged(4),
            DataWatchRemoved(5),
            ChildWatchRemoved(6);
        }
        static enum KeeperState {
            Unknown(-1),
            Disconnected(0),
            NoSyncConnected(1),
            SyncConnected(3),
            AuthFailed(4),
            ConnectedReadOnly(5),
            SaslAuthenticated(6),
            Expired(-112),
            Closed(7);
        }
    }
}

ch := make(chan string)
go func() {
    for i := 0; i < 10; i++ {
        ch <- strconv.Itoa(i) // Отправляем данные в канал
        time.Sleep(1 * time.Second)
    }
}() // Запускаем анонимную горутину

for {
    select { (3)
        case i := <- ch: // Читаем данные из канала
            fmt.Println("Got number: %s", i)
            continue
        case <- closeCh:
        	return
    }
}

// Запросы на чтение
func (client *ZkClient) Exists(path string) (ok bool, stat *zk.Stat, err error)
func (client *ZkClient) Get(path string) (data []byte, stat *zk.Stat, err error)
func (client *ZkClient) Children(path string) (children []string, stat *zk.Stat, err error)

// Запросы с подпиской.
func (client *ZkClient) ExistsW(path string) (ok bool, stat *zk.Stat, evt <-chan zk.Event, err error)
func (client *ZkClient) GetW(path string) (data []byte, stat *zk.Stat, evt <-chan zk.Event, err error)
func (client *ZkClient) ChildrenW(path string) (children []string, stat *zk.Stat, evt <-chan zk.Event, err error)

// Запросы на изменение
func (client *ZkClient) Create(path string, data []byte, flags int32, acl []zk.ACL) (res string, err error)
func (client *ZkClient) Set(path string, data []byte, version int32) (stat *zk.Stat, err error)
func (client *ZkClient) Delete(path string, version int32) (err error)

path := "/some_node"
for {
    data, _, evt, err := cli.GetW(path) // Получение данных из /some_node с подпиской
    if err == zk.ErrNoNode { // Если /some_node не существует, продолжим цикл сначала
        continue
    } else if err != nil {
        panic(err)
    }
    fmt.Printf("Got new value %s", string(data))
    select {
    case e := <-evt:
        if e.Type != zk.EventNodeDataChanged {
            fmt.Printf("Unexpected type of event %s\n", e.Type)
        }
        continue //  Переходим в начало цикла.
    }
}

func (l *Lock) Lock() error {
    path := l.CreateLockNode() //Создаем EPHEMERAL_SEQUENTIAL -lock/guid-00000x
    order := parseSeq(path) // Определяем номер из path znode, созданной на пред. шаге

    for {
        children, _, evt, _ := l.zk_client.ChildrenW(l.path) // Берем дочерние узлы и ставим watch (канал evt)
        lowestOrder := process(children)

        if order == lowestOrder { // Если у нашего узла самый низкий номер, мы захватили лок
            break
        }

        ev := <-evt // Приостанавливаем сопрограмму пока не придет событие из канала evt
        if ev.Err != nil {
            return ev.Err
        }
    }
    l.lock = path
    return nil
}

func (l *Lock) Unlock() {
    _ = l.zk_client.Delete(l.lock, -1) // Отпуская мьютекс, просто удалем znode
}

func (l *Lock) Lock() error {
    path := l.CreateLockNode()
    order := parseSeq(path)

    for {
        children, _, _ := l.zk_client.Children(l.path) // Берем дочерние узлы без отслеживания
        lowestOrder, prevOrderPath := process(children, order)

        if order == lowestOrder {
            break
        }

        exists, _, ch, err := l.zk_client.ExistsW(l.path + "/" + prevOrderPath) // Берем предыдущую по порядку znode
        if err != nil {
            return err
        } else if !exists { // Если она была удалена, переходим в начало цикла
            continue
        }

        ev := <-ch // Иначе приостанавливаем сопрограмму пока не придет событие в канал
        if ev.Err != nil {
            return ev.Err
        }
    }
    l.lock = path
    return nil
}

func NewService() *Service {
    service := &Service{
        ....
    }
    service.localNodePath = createServiceNode() // Создаем znode EPHEMERAL_SEQUENTIAL с именем _services/guid-0000x
    go loop()
    return &service
}

func (service *Service) loop() {
    LOOP:
    for {
        isLeader, changeLeaderChan, err := service.checkLeader(service.localNodePath) // Проверяем кто сейчас лидер.

        if isLeader {
            go service.lead() // Если текущий сервис лидер, запускаем сопрограмму lead
        } else {
            go service.follow() // Иначе сопрограмму follow
        }

        for {
            select {
            case e := <-changeLeaderChan: // Отслеживаем события из канала, на который создали ранее
                switch e.Type {
                case zk.EventNodeDeleted: // Если отслеживаемая znode была удалена, переходим в начало цикла
                    continue LOOP
                default: // Иначе переподписываемся (заодно проверяем что znode не была удалена после события)
                    var exists bool
                    exists, _, changeLeaderChan, _ = service.zk_client.ExistsW(e.Path)

                    if !exists {
                        continue LOOP
                    }
                    continue
                }
            }
        }
    }
}

func (service *Service) checkLeader(nodePath string) (bool, ev <-chan zk.Event, error) {
    for {
    	children, _, err := disco.client.Children(alivePath) // Берем все дочернии узлы без отслеживания
        order := parseSeq(nodePath)
        lowestOrder, previousOrderPath := process(children, order)

        if lowestOrder == order { // Если у текущего процесса самый низкий порядок, он лидер
            return true, nil, nil
        } else {
            exists, _, ev, err = service.zk_client.ExistsW(previousOrderPath) // Иначе отслеживаем предыдущий по порядку
            if !exists {
                continue
            } else {
                return false, ev, nil
            }
        }
    }
}