Using ZooKeeper for distributed coordination in Java API development

With the continuous improvement of computer system performance and the continuous reduction of hardware costs, distributed systems are becoming more and more important in the field of modern computing. Along with this, the demand for distributed computing continues to expand, and the coordination and management solutions for distributed systems become increasingly important.

There are many solutions to achieve distributed coordination, and ZooKeeper is one of the popular solutions. ZooKeeper is one of the sub-projects of the Apache Hadoop project. It provides a reliable distributed coordination service, making it easier for application developers to implement distributed systems.

Using ZooKeeper for distributed coordination in Java API development has become a hot topic. This article will explore some basic concepts of ZooKeeper and provide practical examples to illustrate how to use ZooKeeper for distributed coordination in Java. .

Introduction to ZooKeeper

ZooKeeper is a distributed service designed to coordinate distributed applications. The main goal of ZooKeeper is to provide developers with a relatively simple coordination service so that they can focus on writing applications.

ZooKeeper has the following characteristics:

• ZooKeeper is a distributed service that can be deployed through multiple nodes to provide high availability.
• ZooKeeper is designed as an architecture with one master node and multiple slave nodes. In this structure, the master node is responsible for coordinating and managing slave nodes and ensuring that data is stored securely.
• ZooKeeper tracks the status and changes of nodes by using "ZooKeeper temporary ordered nodes". These nodes are a special type of node that establishes a one-time connection between their creator and the ZooKeeper service. If the connection is lost, the node will be deleted, thus ensuring that the node status is updated in a timely manner.
• ZooKeeper can manage the consistency and integrity of data by using the version control function. When using version control, ZooKeeper will increment the version number of each node.

Basic operations of ZooKeeper

When using ZooKeeper for distributed coordination, the most commonly used operations are creating nodes, reading nodes, and monitoring the status of nodes.

Create Node

To create a node, you need to provide the node path and node data. The node will be added to the ZooKeeper service as a subdirectory. If the node created is an ephemeral node, it can only be accessed as long as the connection between the client that created it and the ZooKeeper service is valid.

The following is sample code for creating a node using the ZooKeeper API:

ZooKeeper zk = new ZooKeeper("localhost:2181", 3000, null);
String nodePath = "/testNode";
byte[] data = "nodeData".getBytes();
CreateMode createMode = CreateMode.EPHEMERAL;
zk.create(nodePath, data, ZooDefs.Ids.OPEN_ACL_UNSAFE, createMode);

Reading a node

You can read and obtain the contents of a node by using the ZooKeeper API. The following is a sample code for reading nodes using the Java API:

ZooKeeper zk = new ZooKeeper("localhost:2181", 3000, null);
String nodePath = "/testNode";
byte[] data = zk.getData(nodePath, false, null);

Monitoring nodes

Monitoring nodes allows the client to be notified of node changes, so that the node status can be updated. The following is a sample code for monitoring nodes using the ZooKeeper API:

ZooKeeper zk = new ZooKeeper("localhost:2181", 3000, null);
String nodePath = "/testNode";
Watcher watcher = new Watcher() {
   public void process(WatchedEvent event) {
      // do something
   }
};
byte[] data = zk.getData(nodePath, watcher, null);

Example of using ZooKeeper for distributed coordination

In the following example, we will implement a simple distributed application using the ZooKeeper API , the application will implement a simple leader election protocol where multiple processes will compete to become the leader. In this case, we will use ZooKeeper ephemeral nodes to implement the leader election functionality.

The following is the sample code:

import java.util.List;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.data.Stat;
 
public class LeaderElection implements Watcher {
    
   String znode = "/leader_election";    

   ZooKeeper zk;
   String serverId = Integer.toHexString((int)(Math.random() * 1000000));
    
   boolean isLeader = false;
    
   public void start() throws Exception{
       
      String serverPath = znode + "/" + serverId;
 
      zk = new ZooKeeper("localhost:2181", 3000, this); 

      while(zk.getState() == ZooKeeper.States.CONNECTING){
         
         Thread.sleep(500); 

      }
       
      while(true){
          
        try{
        
            // create the node with EPHEMERAL and SEQUENTIAL flags
            
            zk.create(serverPath, null, ZooDefs.Ids.OPEN_ACL_UNSAFE,
              CreateMode.EPHEMERAL);
          
            isLeader = true; 

            doLeaderAction();
            break;

         } catch (NodeExistsException e){
                
            isLeader = false;
            break; 

         } catch (InterruptedException e) {
             
             throw e;
             
         } catch (Exception e) {
             
             throw new RuntimeException(e); 
             
         }
      }
   }
    
   public void stop() throws Exception{
       
      zk.close(); 
       
   }
   
   void doLeaderAction() throws Exception {
       
      System.out.println("Becoming leader: " + serverId);
       
      try {            
               
         Thread.sleep(6000);
               
      } catch (InterruptedException e) {

         System.err.println("Interrupted while " +
               "sleeping during leadership.");
         
         Thread.currentThread().interrupt();
      } finally {

         try {               
            System.out.println("Ending leader: " + serverId);
         } catch (Exception e) {
            System.err.println("Error ending leadership."); 
         }
      }
   }
    
   public void process(WatchedEvent e){
       
      System.out.println(e.toString() + ", " + serverId);
      try {
        electLeader();
      } catch (Exception ex) {
        ex.printStackTrace();
      }   
   }
    
   void electLeader() throws Exception {
       
      Stat predecessorStat = null;
      String predecessor = null;
      
      List<String> children = zk.getChildren(znode, false); //(watcher not needed for this operation)
      
      int currentId = Integer.parseInt(serverId, 16); 
       
      for(String child : children){
          
        int childId = Integer.parseInt(child, 16);
        
        if(childId < currentId) {
            
            if(predecessorStat == null){
                
                predecessor = child; 
                predecessorStat = zk.exists(znode + "/" + child, true); 

            } else {
                
                Stat stat = zk.exists(znode + "/" + child, true);
              
                if(stat.getMzxid() < predecessorStat.getMzxid()){
                    
                    predecessor = child; 
                    predecessorStat = stat; 
                }               
            }
        }

      }
       
      if(predecessor == null){
           
        System.out.println("No active group members, " + serverId + " as leader.");
        //...provisional leader code here
           
      } else{ // watch the predecessor node waiting for it to go
                // ...down or to receive a message that it is was elected leader too.        
        System.out.println("Watching group member with higher ID: " + predecessor);
      }         
   }
   
   public static void main(String[] args) throws Exception {
          
      LeaderElection election = new LeaderElection();
      
      election.start();
       
   }
}

In the above sample code, we first create a znode subdirectory that is used to maintain the participation status of all processes participating in leader election. Next, we create a temporary ordered ZooKeeper node that represents a given actor. As mentioned before, ZooKeeper establishes a one-time connection between the client and the Zk value. After we create this temporary node, if the client connection is lost, the node will be deleted. Therefore, if a process discovers that a node with the same node name already exists when establishing a node, the process will not become the leader.

If the client successfully creates the temporary node, the client will become the leader. Here we can call the doLeaderAction() method which represents the action the leader will perform. In this example, the leader will perform a simple 6-second operation.

If the client connection has been lost or any error occurs, the process verifies the nodes under the existing directory to determine which one becomes the new leader.

Conclusion

Distributed coordination and management is one of the most important issues in the field of modern computing, and the application of distributed systems is becoming more and more popular. ZooKeeper is a popular solution that makes it easier for developers to implement distributed systems. In Java API development, the main operations of using ZooKeeper for distributed coordination include creating nodes, reading nodes, and monitoring the status of nodes. Through the sample code in this article, you can see how to use ZooKeeper to implement the leader election protocol and other distributed coordination schemes in Java.

The above is the detailed content of Using ZooKeeper for distributed coordination in Java API development. For more information, please follow other related articles on the PHP Chinese website!