ZK注册中心-思路

一、思路

服务方启动则将自己注册到zk上，临时节点，节点数据为IP和端口信息
客户端添加监听器，监听节点变化，每次变化更新本地服务列表
服务端有问题，则自动摘除节点，依靠临时节点实现

二、注册方实现

1）添加节点监听器

public class ServiceRegister {
    private  static final String BASE_SERVICES = "/services";
    private static final String  SERVICE_NAME="/products";
    public  static void register(String address,int port) {
        try {
            //产品服务最终会注册倒的地址
            String path = BASE_SERVICES + SERVICE_NAME;
            ZooKeeper zooKeeper =  new ZooKeeper("192.168.112.131:2181",5000, (watchedEvent)->{});


            createNodeSafe(   zooKeeper,BASE_SERVICES);
            createNodeSafe( zooKeeper ,path );
            //拼接ip和端口
            String server_path = address+":" +port;
            //注册的类型，EPHEMERAL_SEQUENTIAL 零时，并且带序号
            zooKeeper.create(path+"/child",server_path.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,CreateMode.EPHEMERAL_SEQUENTIAL);
            System.out.println("产品服务注册成功");


        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    private static void createNodeSafe(ZooKeeper zooKeeper , String path) throws KeeperException, InterruptedException {
        Stat exists = zooKeeper.exists(BASE_SERVICES + SERVICE_NAME, false);
        //先判断服务根路径是否存在
        if(exists == null) {
            zooKeeper.create(path,"".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }
    }
}

2）创建一个监听器

public class InitListener implements ServletContextListener {
    @Override
    //容器初始化的时候会调用，重启tomcat
    public void contextInitialized(ServletContextEvent sce) {
            try {
                Properties properties =  new Properties();
                properties.load(InitListener.class.getClassLoader().getResourceAsStream("application.properties"));
                //获得IP
                String hostAddress = InetAddress.getLocalHost().getHostAddress();
                //获得端口
                int port = Integer.valueOf(properties.getProperty("server.port"));
                ServiceRegister.register(hostAddress,port);
            } catch (Exception e) {
                e.printStackTrace();
            }
    }
    @Override
    public void contextDestroyed(ServletContextEvent sce) {


    }
}

3）注册监听器，容器启动的时候加载

@Bean
//启动监听起，当tomcat启动的时候，会调用InitListener
public ServletListenerRegistrationBean servletListenerRegistrationBean() {
    ServletListenerRegistrationBean servletListenerRegistrationBean = new ServletListenerRegistrationBean();
    servletListenerRegistrationBean.setListener(new InitListener());
    return  servletListenerRegistrationBean;
}

三、使用方

1）服务启动添加节点监听器

public class InitListener implements ServletContextListener {
    private  static final String BASE_SERVICES = "/services";
    private static final String  SERVICE_NAME="/products";
    private  static AtomicInteger errorTryTimes = new AtomicInteger(0);
    private static final String zkAddr ="192.168.112.131:2181" ;
    private static final Integer zkTimeout = 5000 ;
    private   ZooKeeper zooKeeper;
    private   void init()  {
        try {
            //连接上zk 获得列表信息
            zooKeeper = new ZooKeeper(zkAddr ,zkTimeout,(watchedEvent)->{
                  //当有节点变更的时候通知倒orderService
                if(watchedEvent.getType() == Watcher.Event.EventType.NodeChildrenChanged
                        && watchedEvent.getPath().equals(BASE_SERVICES+SERVICE_NAME)) {
                    updateServerList();
                }
            });

            //第一次连接的时候要获得列表
            updateServerList();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private  void updateServerList() {
        List<String> newServerList = new ArrayList<>();
        try {
            List<String> children = zooKeeper.getChildren(BASE_SERVICES  + SERVICE_NAME, true);
            for(String subNode:children) {
                byte[] data = zooKeeper.getData(BASE_SERVICES  + SERVICE_NAME + "/" + subNode, false, null);
                String host = new String(data, "utf-8");
                System.out.println("host:"+host);
                newServerList.add(host);
            }
            LoadBalance.SERVICE_LIST = newServerList;
        } catch (Exception e) {
            //可能异常导致本地列表没有更新，则异常可以重试
            if(errorTryTimes.incrementAndGet()<3){
                init();
            }
            //打印日志
            e.printStackTrace();
        }
    }

    @Override
    public void contextInitialized(ServletContextEvent sce) {
        init();
    }
    @Override
    public void contextDestroyed(ServletContextEvent sce) {
    }
}

View Code

注册监听器：

@Bean
public ServletListenerRegistrationBean servletListenerRegistrationBean() {
    ServletListenerRegistrationBean servletListenerRegistrationBean = new ServletListenerRegistrationBean();
    servletListenerRegistrationBean.setListener(new InitListener());
    return  servletListenerRegistrationBean;
}

2）客户端保存服务端的服务列表数据

public abstract class LoadBalance {
    //最好使用set，这样如果有重复的直接去掉
    public volatile static   List<String> SERVICE_LIST;
    public abstract String choseServiceHost();
}

3）简单的负载均衡，访问可以通过随机获得的ip端口，然后拼接对应的参数访问：http或者其他的都可以

public class RamdomLoadBalance extends  LoadBalance {
    @Override
    public String choseServiceHost() {
        String result = "";
        //SERVICE_LIST 产品 服务的列表
        if(!CollectionUtils.isEmpty(SERVICE_LIST)) {
           // 192.168.30.3:8083
            //192.168.30.3:8084  2
            //index 0,1
            //传入一个种子
            int index = new Random().nextInt(SERVICE_LIST.size());
            result = SERVICE_LIST.get(index);
        }
        return result ;
    }
}