ZooKeeper 筆記(2) 監聽數據變化 - yjmyzz－IT工程師數位筆記本

文章出處

ZK中的每個節點都可以存儲一些輕量級的數據，這些數據的變化會同步到集群中的其它機器。在應用中程序員可以添加watcher來監聽這些數據的變化，watcher只會觸發一次，所以觸發過后想要繼續監聽，必須再手動設置監聽，這比較麻煩，好在ZkClient已經做了一些增強，在watcher的基礎上，封裝了所謂的Listener，開發人員只需要訂閱Listener即可。

package yjmyzz.test;

import com.cnblogs.yjmyzz.domain.BaseBean;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import java.util.concurrent.TimeUnit;


public class ZKTest extends BaseBean {

    private ZkClient zk;

    private String nodeName = "/myApp";

    @Before
    public void initTest() {
        super.setLoggerClass(this.getClass());
        zk = new ZkClient("localhost:2181,localhost:2182,localhost:2183");
    }

    @After
    public void dispose() {
        zk.close();
        logger.info("zkclient closed!");
    }

    @Test
    public void testListener() throws InterruptedException {
        //監聽指定節點的數據變化

        zk.subscribeDataChanges(nodeName, new IZkDataListener() {
            @Override
            public void handleDataChange(String s, Object o) throws Exception {
                logger.info("node data changed!");
                logger.info("node=>" + s);
                logger.info("data=>" + o);
                logger.info("--------------");
            }

            @Override
            public void handleDataDeleted(String s) throws Exception {
                logger.info("node data deleted!");
                logger.info("s=>" + s);
                logger.info("--------------");

            }
        });

        logger.info("ready!");

        //junit測試時，防止線程退出
        while (true) {
            TimeUnit.SECONDS.sleep(5);
        }
    }


    @Test
    public void testUpdateConfig() throws InterruptedException {
        if (!zk.exists(nodeName)) {
            zk.createPersistent(nodeName);
        }
        zk.writeData(nodeName, "1");
        zk.writeData(nodeName, "2");
        zk.delete(nodeName);
        zk.delete(nodeName);//刪除一個不存在的node，并不會報錯
    }

}

上面的代碼中，演示了如何監聽/myApp的數據變化，用junit做單元測試時，步驟如下：

1. 先測試testListener()方法，由于這個方法最后，寫了一個死循環，所以程序不會退出。

2. 緊接著再測試testUpdateConfig()，這個方法中修改了節點的數據。

此時，由于testListener中設置了監聽，所以監聽程序應該會起作用，打印出相應的數據變化，類似下面的效果：

yjmyzz.test.ZKTest - ready!
yjmyzz.test.ZKTest - node data changed!
yjmyzz.test.ZKTest - node=>/myApp
yjmyzz.test.ZKTest - data=>1
yjmyzz.test.ZKTest - --------------
yjmyzz.test.ZKTest - node data changed!
yjmyzz.test.ZKTest - node=>/myApp
yjmyzz.test.ZKTest - data=>2
yjmyzz.test.ZKTest - --------------
yjmyzz.test.ZKTest - node data deleted!
yjmyzz.test.ZKTest - s=>/myApp
yjmyzz.test.ZKTest - --------------
yjmyzz.test.ZKTest - node data deleted!
yjmyzz.test.ZKTest - s=>/myApp
yjmyzz.test.ZKTest - --------------