http://blog.csdn.net/ZuoAnYinXiang/article/category/6104448

 

1.发布订阅的基本概念

 

     1.发布订阅模式可以看成一对多的关系：多个订阅者对象同时监听一个主题对象，这个主题对象在自身状态发生变化时，会通知所有的订阅者对象，使他们能够自动的更新自己的状态。

 

     2.发布订阅模式，可以让发布方和订阅方，独立封装，独立改变，当一个对象的改变，需要同时改变其他的对象，而且它不知道有多少个对象需要改变时，可以使用发布订阅模式

 

    3.发布订阅模式在分布式系统的典型应用有， 配置管理和服务发现。

           配置管理：是指如果集群中机器拥有某些相同的配置，并且这些配置信息需要动态的改变，我们可以使用发布订阅模式，对配置文件做统一的管理，让这些机器各       自订阅配置文件的改变，当配置文件发生改变的时候这些机器就会得到通知，把自己的配置文件更新为最新的配置

    

            服务发现：是指对集群中的服务上下线做统一的管理，每个工作服务器都可以作为数据的发布方，向集群注册自己的基本信息，而让模型机器作为订阅方，订阅工       作服务器的基本信息，当工作服务器的基本信息发生改变时如上下线，服务器的角色和服务范围变更，监控服务器就会得到通知，并响应这些变化。

 

2.发布订阅的图

3.Manager Server的工作流程

4.Work Server的工作流程

5.发布订阅程序的结构图

 

 

 

 

6.程序代码实现

 

package com.zk.subscribe;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.serialize.BytesPushThroughSerializer;


public class SubscribeZkClient {

        //需要多少个workserver
        private static final int  CLIENT_QTY = 5;

        private static final String  ZOOKEEPER_SERVER = "192.168.30.164:2181,192.168.30.165:2181,192.168.30.166:2181";
        //节点的路径
        private static final String  CONFIG_PATH = "/config";//配置节点
        private static final String  COMMAND_PATH = "/command";//命令节点
        private static final String  SERVERS_PATH = "/servers";//服务器列表节点

        public static void main(String[] args) throws Exception
        {
            //用来存储所有的clients
            List<ZkClient>  clients = new ArrayList<ZkClient>();
            //用来存储所有的workservers
            List<WorkServer>  workServers = new ArrayList<WorkServer>();
            ManagerServer manageServer = null;

            try
            {
                ServerConfig initConfig = new ServerConfig();
                initConfig.setDbPwd("123456");
                initConfig.setDbUrl("jdbc:mysql://localhost:3306/mydb");
                initConfig.setDbUser("root");

                ZkClient clientManage = new ZkClient(ZOOKEEPER_SERVER, 5000, 5000, new BytesPushThroughSerializer());
                manageServer = new ManagerServer(SERVERS_PATH, COMMAND_PATH,CONFIG_PATH,clientManage,initConfig);
                manageServer.start();

                //根据定义的work服务个数，创建服务器后注册，然后启动
                for ( int i = 0; i < CLIENT_QTY; ++i )
                {
                    ZkClient client = new ZkClient(ZOOKEEPER_SERVER, 5000, 5000, new BytesPushThroughSerializer());
                    clients.add(client);
                    ServerData serverData = new ServerData();
                    serverData.setId(i);
                    serverData.setName("WorkServer#"+i);
                    serverData.setAddress("192.168.1."+i);

                    WorkServer  workServer = new WorkServer(CONFIG_PATH, SERVERS_PATH, serverData, client, initConfig);
                    workServers.add(workServer);
                    workServer.start();

                }
                System.out.println("敲回车键退出！\n");
                new BufferedReader(new InputStreamReader(System.in)).readLine();

            }finally{
                //将workserver和client给关闭

                System.out.println("Shutting down...");

                for ( WorkServer workServer : workServers )
                {
                    try {
                        workServer.stop();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                for ( ZkClient client : clients )
                {
                    try {
                        client.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }
            }
        }

}

 

package com.zk.subscribe;

import java.util.List;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import com.alibaba.fastjson.JSON;

public class ManagerServer {

    private String serversPath;
    private String commandPath;
    private String configPath;
    private ZkClient zkClient;
    private ServerConfig config;
    //用于监听zookeeper中servers节点的子节点列表变化
    private IZkChildListener childListener;
    //用于监听zookeeper中command节点的数据变化
    private IZkDataListener dataListener;
    //工作服务器的列表
    private List<String> workServerList;

    /**
     *
     * @param serversPath
     * @param commandPath Zookeeper中存放命令的节点路径
     * @param configPath
     * @param zkClient
     * @param config
     */
    public ManagerServer(String serversPath, String commandPath,String configPath, ZkClient zkClient, ServerConfig config) {
        this.serversPath = serversPath;
        this.commandPath = commandPath;
        this.zkClient = zkClient;
        this.config = config;
        this.configPath = configPath;
        this.childListener = new IZkChildListener() {
            //用于监听zookeeper中servers节点的子节点列表变化
            public void handleChildChange(String parentPath,List<String> currentChilds) throws Exception {
                //更新服务器列表
                workServerList = currentChilds;

                System.out.println("work server list changed, new list is ");
                execList();

            }
        };

        //用于监听zookeeper中command节点的数据变化
        this.dataListener = new IZkDataListener() {

            public void handleDataDeleted(String dataPath) throws Exception {

            }

            public void handleDataChange(String dataPath, Object data)
                    throws Exception {

                String cmd = new String((byte[]) data);
                System.out.println("cmd:"+cmd);
                exeCmd(cmd);

            }
        };

    }

    public void start() {
        initRunning();
    }

    public void stop() {
        //取消订阅command节点数据变化和servers节点的列表变化
        zkClient.unsubscribeChildChanges(serversPath, childListener);
        zkClient.unsubscribeDataChanges(commandPath, dataListener);
    }

    /**
     * 初始化
     */
    private void initRunning() {
        //执行订阅command节点数据变化和servers节点的列表变化
        zkClient.subscribeDataChanges(commandPath, dataListener);
        zkClient.subscribeChildChanges(serversPath, childListener);

    }


    /*
     * 执行控制命令的函数
     * 1: list 2: create 3: modify
     */
    private void exeCmd(String cmdType) {
        if ("list".equals(cmdType)) {
            execList();

        } else if ("create".equals(cmdType)) {
            execCreate();
        } else if ("modify".equals(cmdType)) {
            execModify();
        } else {
            System.out.println("error command!" + cmdType);
        }

    }


    private void execList() {

        System.out.println(workServerList.toString());
    }

    private void execCreate() {
        if (!zkClient.exists(configPath)) {
            try {

                zkClient.createPersistent(configPath, JSON.toJSONString(config).getBytes());

            } catch (ZkNodeExistsException e) {
                //节点已经存在异常，直接写入数据
                zkClient.writeData(configPath, JSON.toJSONString(config).getBytes());
            } catch (ZkNoNodeException e) {
                //表示其中的一个节点的父节点还没有被创建
                String parentDir = configPath.substring(0,configPath.lastIndexOf('/'));
                zkClient.createPersistent(parentDir, true);
                execCreate();

            }
        }
    }

    private void execModify() {
        config.setDbUser(config.getDbUser() + "_modify");

        try {
            //回写到zookeeper中
            zkClient.writeData(configPath, JSON.toJSONString(config).getBytes());
        } catch (ZkNoNodeException e) {
            execCreate();
        }
    }


}

 

package com.zk.subscribe;
import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.I0Itec.zkclient.exception.ZkNodeExistsException;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;

/**
 * 代表工作服务器
 * workServer服务器的信息
 *
 */
public class WorkServer{
    private String serversPath;
    private String configPath;
    private ZkClient zkClient;
    private ServerConfig config;
    private ServerData serverData;

    private IZkDataListener dataListener;//数据监听器


    /**
     *
     * @param configPath 代表config节点的路径
     * @param serversPath 代表servers节点的路径
     * @param serverData   代表当前服务器的基本信息
     * @param zkClient     底层与zookeeper集群通信的组件
     * @param initconfig   当前服务器的初始配置
     */
    public WorkServer(String configPath,String serversPath,ServerData serverData,ZkClient zkClient, ServerConfig initconfig){

        this.configPath = configPath;
        this.serversPath = serversPath;
        this.serverData = serverData;
        this.zkClient = zkClient;
        this.config = initconfig;

        /**
         * dataListener 用于监听config节点的数据改变
         */
        this.dataListener = new IZkDataListener() {

            public void handleDataDeleted(String arg0) throws Exception {

            }

            /**
             * 当数据的值改变时处理的
             * Object data,这个data是将ServerConfig对象转成json字符串存入
             * 可以通过参数中的Object data 拿到当前数据节点最新的配置信息
             * 拿到这个data信息后将它反序列化成ServerConfig对象，然后更新到自己的serverconfig属性中
             */
            public void handleDataChange(String dataPath, Object data) throws Exception {
                String retJson = new String((byte[])data);
                ServerConfig serverConfigLocal = (ServerConfig)JSON.parseObject(retJson,ServerConfig.class);
                //更新配置
                updateConfig(serverConfigLocal);
                System.out.println("new work server config is:"+serverConfigLocal.toString());
            }
        };

    }

    /**
     * 服务的启动
     */
    public void start(){
        System.out.println("work server start...");
        initRunning();
    }

    /**
     * 服务的停止
     */
    public void stop(){
        System.out.println("work server stop...");
        //取消监听
        zkClient.unsubscribeDataChanges(configPath, dataListener);

    }

    /**
     * 服务器的初始化
     */
    private void initRunning(){
        registMeToZookeeper();
        //订阅config节点的改变
        zkClient.subscribeDataChanges(configPath, dataListener);
    }

    /**
     * 启动时向zookeeper注册自己
     */
    private void registMeToZookeeper(){
        //向zookeeper中注册自己的过程其实就是向servers节点下注册一个临时节点
        //构造临时节点
        String mePath = serversPath.concat("/").concat(serverData.getAddress());
        try{
            //存入是将json序列化
            zkClient.createEphemeral(mePath, JSON.toJSONString(serverData).getBytes());
        } catch (ZkNoNodeException e) {
            //父节点不存在
            zkClient.createPersistent(serversPath, true);
            registMeToZookeeper();
        }

    }

    /**
     * 当监听到zookeeper中config节点的配置信息改变时，要读取配置信息来更新自己的配置信息
     */
    private void updateConfig(ServerConfig serverConfig){
        this.config = serverConfig;
    }

}

 

package com.zk.subscribe;

/**
 * 用于记录WorkServer（工作服务器）的配置信息
 */
public class ServerConfig {
    private String dbUrl;
    private String dbPwd;
    private String dbUser;
    public String getDbUrl() {
        return dbUrl;
    }
    public void setDbUrl(String dbUrl) {
        this.dbUrl = dbUrl;
    }
    public String getDbPwd() {
        return dbPwd;
    }
    public void setDbPwd(String dbPwd) {
        this.dbPwd = dbPwd;
    }
    public String getDbUser() {
        return dbUser;
    }
    public void setDbUser(String dbUser) {
        this.dbUser = dbUser;
    }
    @Override
    public String toString() {
        return "ServerConfig [dbUrl=" + dbUrl + ", dbPwd=" + dbPwd + ", dbUser=" + dbUser + "]";
    }


}

 

package com.zk.subscribe;
/**
 * 用于记录WorkServer（工作服务器）的基本信息
 */
public class ServerData {
    private String address;
    private Integer id;
    private String name;
    public String getAddress() {
        return address;
    }
    public void setAddress(String address) {
        this.address = address;
    }
    public Integer getId() {
        return id;
    }
    public void setId(Integer id) {
        this.id = id;
    }
    public String getName() {
        return name;
    }
    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "ServerData [address=" + address + ", id=" + id + ", name=" + name + "]";
    }


}

 

启动SubscribeZkClient

 

在zookeeper客户端上输出命令

 

managerServer订阅了commod的变化后，输出变化