• 发布/订阅模式是一对多的关系，多个订阅者对象同时监听某一主题对象，这个主题对象在自身状态发生变化时会通知所有的订阅者对象。使它们能自动的更新自己的状态。发布/订阅可以使得发布方和订阅方独立封装、独立改变。当一个对象的改变需要同时改变其他对象，而且它不知道具体有多少对象需要改变时可以使用发布/订阅模式。发布/订阅模式在分布式系统中的典型应用有配置管理和服务发现与注册。
• 配置管理是指如果集群中的机器拥有某些相同的配置并且这些配置信息需要动态的改变，我们可以使用发布/订阅模式把配置做统一集中管理，让这些机器格子各自订阅配置信息的改变，当配置发生改变时，这些机器就可以得到通知并更新为最新的配置。
• 服务发现于注册是指对集群中的服务上下线做统一管理。每个工作服务器都可以作为数据的发布方向集群注册自己的基本信息，而让某些监控服务器作为订阅方，订阅工作服务器的基本信息，当工作服务器的基本信息发生改变如上下线、服务器角色或服务范围变更，监控服务器可以得到通知并响应这些变化。

1、发布订阅模式讲解

• 发布/订阅系统一般有两种设计模式，分别是推(Push)模式和拉(Pull)模式。

推模式:服务端主动将数据更新发送给所有订阅的客户端
拉模式：客户端主动请求获取最新数据，通常客户端都采用定时进行轮询拉取的方式。

• ZooKeeper采用的是推拉相结合的方式：客户端向服务端注册自己需要关注的节点，一旦该节点的数据发生变更，那么服务端就会向相应的客户端发送Watcher事件通知，客户端接收到这个消息通知之后，需要主动到服务端获取最新的数据。

1、配置管理

• 所谓基于zk的配置中心，顾名思义就是发布者将数据发布到 ZooKeeper 的一个或一系列节点上，供订阅者进行数据订阅，进而达到动态获取数据的目的，实现配置信息的集中式管理和数据的动态更新。
• 前面知道了zk采用推拉相结合的方式进行数据的发布和订阅的。如果将配置信息存放到ZK上进行集中管理，那么通常情况下，应用在启动的时候会主动到ZK服务器上进行一次配置信息的获取，同时，在指定上注册一个Watcher监听，这样一来，但凡配置信息发生变更，服务器都会实时通知所有订阅的客户端，从而达到实时获取最新配置信息的目的。

（1）案例分析

• 案例场景：在我们平常的应用系统开发中，经常会碰到这样的需求：系统中需要使用一些通用的配置信息，例如机器列表信息、运行时的开关配置、数据库的配置信息等。这些全局配置信息通常具备以下特性：

1)、数据量通常比较小
2)、数据内容在运行时会发生变化
3)、集群中各机器共享、配置一致

对于这类配置信息，一般的做法通常可以选择将其存储的本地配置文件或是内存变量中。无论采取哪种配置都可以实现相应的操作。但是一旦遇到集群规模比较大的情况的话，两种方式就不再可取。而我们还需要能够快速的做到全部配置信息的变更，同时希望变更成本足够小，因此我们需要一种更为分布式的解决方案。此时就可以使用zk来解决。

• zk实现配置管理原理：

1、配置存储：在进行配置管理之前，首先我们需要将初始化配置存储到ZK上去，一般情况下，我们可以在ZK上选取一个数据节点用于配置的存储，例如/app1/database_config（配置节点），我们将需要集中管理的配置信息写入到该数据节点中去。
2、配置获取：集群中每台机器在启动初始化阶段，首先会从上面提到的ZK的配置节点上读取数据库信息，同时，客户端还需要在该配置节点上注册一个数据变更的Watcher监听，一旦发生节点数据变更，所有订阅的客户端都能够获取数据变更通知。　　
3、配置变更：在系统运行过程中，可能会出现需要进行全局切换的情况，这个时候就需要进行配置变更。借助ZK，我们只需要对ZK上配置节点的内容进行更新，ZK就能够帮我们将数据变更的通知发送到各个客户端，每个客户端在接收到这个变更通知后，就可以重新进行最新数据的获取。

（2）案例demo

/**
 * 下面demo就是一个典型的发布订阅系统：
 * 集群中每台机器在启动阶段，都会到该节点上获取数据库的配置信息，同时客户端还需要在在节
 * 点注册一个数据变更的watcher监听，一旦该数据节点发生变更，就会受到通知信息。
 */
public class ConfigTest {
    /**
     * 配置中心父节点
     */
    private static final String PATH = "/server/database_config";
    private static final String zkAddress = "127.0.0.1:2181";
    private static final int timeout = 1000;

    private static CuratorFramework client = null;

    private static CountDownLatch countDownLatch = new CountDownLatch(1);

    /**
     * 客户端的连接状态监听
     */
    static ConnectionStateListener clientListener = new ConnectionStateListener() {

        public void stateChanged(CuratorFramework client,
                                 ConnectionState newState) {
            if (newState == ConnectionState.CONNECTED) {
                System.out.println("connected established");
                countDownLatch.countDown();
            } else if (newState == ConnectionState.LOST) {
                System.out.println("connection lost,waiting for reconection");
                try {
                    System.out.println("reinit---");
                    reinit();
                    System.out.println("inited---");
                } catch (Exception e) {
                    System.err.println("re-inited failed");
                }
            } else if (newState == ConnectionState.SUSPENDED) {
                System.out.println("suspended");
            } else {
                System.out.println(newState);
            }

        }
    };

    public static void main(String[] args) throws Exception {
        //1、初始化curator
        init();
        //2、判断父节点是否存在，不在的话创建该节点（这个节点本身应在服务器手动添加的）
        Stat stat = client.checkExists().forPath(PATH);
        if (stat == null) {
            client.create().creatingParentsIfNeeded().forPath(PATH);
        }
        //3、对path的变更进行监听
        watcherPath(PATH, pathWatcher);
        //4、模拟阻塞场景，可以客户端改变数据测试
        Thread.sleep(Integer.MAX_VALUE);
    }


    public static void init() throws Exception {
        client = CuratorFrameworkFactory.builder().connectString(zkAddress)
                .sessionTimeoutMs(timeout)
                .retryPolicy(new RetryNTimes(5, 5000)).build();
        // 客户端注册连接状态监听器，进行连接配置（客户端连接的状态会被相应 的监听器监听）
        client.getConnectionStateListenable().addListener(clientListener);
        client.start();
        // 连接成功后，才进行下一步的操作（连接成功会触发监听器中的countDownLatch.await()）
        countDownLatch.await();
    }

    public static void reinit() {
        try {
            unregister();
            init();
        } catch (Exception e) {
            // TODO: handle exception
        }
    }

    public static void unregister() {
        try {
            if (client != null) {
                client.close();
                client = null;
            }
        } catch (Exception e) {
            System.out.println("unregister failed");
        }
    }

    /**
     * 对path进行监听配置
     *
     * @param path
     * @param watcher
     * @return
     * @throws Exception
     */
    public static String watcherPath(String path, CuratorWatcher watcher)
            throws Exception {
        //只是改变数据时候会反应（或者是使用getChildren、exist。或者是在创建客户端构造函数进行watcher监听）
        byte[] buffer = client.getData().usingWatcher(watcher).forPath(path);

        System.out.println("获取节点的信息:" + new String(buffer));
        return new String(buffer);
    }

    /**
     * 读取path数据
     *
     * @param path
     * @return
     * @throws Exception
     */
    public static String readPath(String path) throws Exception {
        byte[] buffer = client.getData().forPath(path);
        return new String(buffer);

    }

    /**
     *
     * 对path进行改变监听的watcher
     */
    private static CuratorWatcher pathWatcher = new CuratorWatcher() {
        public void process(WatchedEvent event) throws Exception {
            System.out.println(event.getType());
            // 当数据变化后，重新获取数据信息
            if (event.getType() == Watcher.Event.EventType.NodeDataChanged) {
                //获取更改后的数据，进行相应的业务处理
                String value = readPath(event.getPath());
                System.out.println(value);
            }

        }
    };

}

测试时候可以先启动main函数，会一直阻塞等待连连接进行。启动zk并连接成功后会触发连接监听器，打印信息。然后会执行注册watcher逻辑，注册之后，sleep模拟生产环境机器的一直运行。当在zk客户端进行数据节点数据的变更时候，会触发pathWatcher中process方法的执行。

2、服务注册与发现

后续看dubbo的时候在回头来实现。