文档章节

【zookeeper】4、利用zookeeper,借助观察模式,判断服务器的上下线

o
 osc_w9s1w4o0
发布于 2019/04/08 10:06
字数 2222
阅读 5
收藏 0

精选30+云产品,助力企业轻松上云!>>>

 首先什么是观察者模式,可以看看我之前的设计模式的文章

https://www.cnblogs.com/cutter-point/p/5249780.html

 

确定一下,要有观察者,要有被观察者,然后要被观察者触发事件,事件发生之后,观察者触发相应的事件发生

了解了基本概念,我们来看看zookeeper是什么情况

zookeeper也是类似观察者一样,我们先把本机信息注册进入服务器,然后设置一个watch方法,这个在zookeeper节点发生变化的时候通知对应的客户端,触发对应的方法

这里先注册服务,如何向zookeeper进行注册呢

 

package cn.cutter.demo.hadoop.zookeeper;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

/**
 * @ProjectName: cutter-point
 * @Package: cn.cutter.demo.hadoop.zookeeper
 * @ClassName: TimeQueryServer
 * @Author: xiaof
 * @Description: 利用zookeeper来进行分布式时间查询
 * @Date: 2019/4/2 19:37
 * @Version: 1.0
 */
public class TimeQueryServer {

    private ZooKeeper zooKeeper;

    // 构造zk客户端连接
    public void connectZK() throws Exception{
        zooKeeper = new ZooKeeper("192.168.1.4:2181,192.168.1.4:2182,192.168.1.4:2183", 2000, null);
    }


    // 注册服务器信息
    public void registerServerInfo(String hostname,String port) throws Exception{

        /**
         * 先判断注册节点的父节点是否存在,如果不存在,则创建
         */
        Stat stat = zooKeeper.exists("/servers", false);
        if(stat==null){
            zooKeeper.create("/servers", null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
        }

        // 注册服务器数据到zk的约定注册节点下
        String create = zooKeeper.create("/servers/server", (hostname+":"+port).getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);

        System.out.println(hostname+" 服务器向zk注册信息成功,注册的节点为:" + create);

    }

}

 

如果注入了服务,那么我们为了监控这个服务的存在,那么是不是应该也模拟一个服务?

好,这里我们就做一个时钟同步的服务,用消费线程不断请求服务,并获取当前时间

package cn.cutter.demo.hadoop.zookeeper;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import java.io.IOException;
import java.io.InputStream;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.*;
import java.util.Date;
import java.util.Iterator;

/**
 * @ProjectName: cutter-point
 * @Package: cn.cutter.demo.hadoop.zookeeper
 * @ClassName: TimeQueryService
 * @Author: xiaof
 * @Description: ${description}
 * @Date: 2019/4/2 19:43
 * @Version: 1.0
 */
public class TimeQueryService extends Thread {

    private static final Log log = LogFactory.getLog(TimeQueryService.class);

    int port = 0;

    public TimeQueryService(int port) {
        this.port = port;
    }

    @Override
    public void run() {

        //1.创建信道选择器
        Selector selector = null;
        //不断读取字符,只有读到换行我们才进行输出
//        StringBuffer stringBuffer = new StringBuffer();
        try {
            selector = Selector.open();
            //2.创建对应端口的监听
            //2.1 创建通道
            ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
            //2.2 socket 对象绑定端口  socket() 获取与此通道关联的服务器套接字
            serverSocketChannel.socket().bind(new InetSocketAddress(port));
            //2.3 设置为非阻塞
            serverSocketChannel.configureBlocking(false);
            //注册到对应的选择器,读取信息
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
        } catch (IOException e) {
            e.printStackTrace();
        }
        //3.轮询获取信息
        while (true) {
            //获取socket对象
            //获取准备好的信道总数
            if (!selector.isOpen()) {
                System.out.println("is close over");
                break;
            }

            try {
                if (selector.select(3000) == 0) {
                    continue; //下一次循环
                }
            } catch (IOException e) {
                e.printStackTrace();
            }

            //获取信道
            Iterator<SelectionKey> keyIterable = selector.selectedKeys().iterator();
            while (keyIterable.hasNext()) {
                //6.遍历键集,判断键类型,执行相应的操作
                SelectionKey selectionKey = keyIterable.next();
                //判断键类型,执行相应操作
                if (selectionKey.isAcceptable()) {
                    try {
                        //从key中获取对应信道
                        //接受数据
                        SocketChannel socketChannel = ((ServerSocketChannel) selectionKey.channel()).accept();
                        //并设置成非阻塞
                        socketChannel.configureBlocking(false);
                        //从新注册,修改状态
                        socketChannel.register(selectionKey.selector(), SelectionKey.OP_READ, ByteBuffer.allocate(1024));
                    } catch (IOException e) {
                        e.printStackTrace();
                    }
                }

                if (selectionKey.isReadable()) {
                    //读取数据
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    //获取当前的附加对象。
                    ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();

                    //判断是是否断开连接
                    int count = 0;
                    while (true) {
                        try {
                            if (!((count = socketChannel.read(byteBuffer)) != 0 && count != -1 && selectionKey.isValid())) {
                                if(count == -1) {
                                    //关闭通道
                                    socketChannel.close();
                                }
                                break;
                            }
                        } catch (IOException e) {
//                                e.printStackTrace();
                            try {
                                //如果读取数据会抛出异常,那么就断定通道已经被客户端关闭
                                socketChannel.close();
                            } catch (IOException e1) {
                                e1.printStackTrace();
                            }
                            System.out.println("无法读取数据!");
                            break;
                        }
                        //判断是否有换行
//                            byteBuffer.flip();
                        byte msg[] = byteBuffer.array();
                        boolean isOver = false;
                        int i = byteBuffer.position() - count;
                        for (; i < byteBuffer.position(); ++i) {
                            //判断是否有换行
                            if (byteBuffer.get(i) == '\r' || byteBuffer.get(i) == '\n') {
                                //输出
                                //先压缩数据
                                byteBuffer.flip();
                                byte out[] = new byte[byteBuffer.limit()];
                                byteBuffer.get(out, 0, out.length);
                                log.info(new String(out));
                                //设置成可以读和可写状态
                                byteBuffer.compact();
                                byteBuffer.clear();
                                isOver = true;
                            }
                        }
                        if (isOver == true) {
//                            interestOps(SelectionKey.OP_READ);的意思其实就是用同一个KEY重新注册
                            selectionKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                            break;
                        }
                    }

                    if (count == -1) {
                        //如果是-1 ,那么就关闭客户端
                        try {
                            socketChannel.close();
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    } else {
//                            selectionKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
                    }
                }

                //告知此键是否有效。
                if (selectionKey.isValid() && selectionKey.isWritable()) {
                    //获取当前的附加对象。
//                        ByteBuffer byteBuffer = (ByteBuffer) selectionKey.attachment();
                    // 清空,并写入数据
//                        byteBuffer.clear();
                    byte smsBytes[] = (new Date().toString() + "\n").getBytes();
                    ByteBuffer byteBuffer = ByteBuffer.wrap(smsBytes);
//                        byteBuffer.put(smsBytes);
                    SocketChannel socketChannel = (SocketChannel) selectionKey.channel();
                    //写入数据
//                        System.out.println(new String(byteBuffer.array()));
                    while (byteBuffer.hasRemaining()) {
                        //输出数据
                        try {
                            socketChannel.write(byteBuffer);
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                    //判断是否判断是否有待处理数据
                    if (!byteBuffer.hasRemaining()) {
                        //数据清理干净
                        selectionKey.interestOps(SelectionKey.OP_READ);
                    }
                    //压缩此缓冲区将缓冲区的当前位置和界限之间的字节(如果有)复制到缓冲区的开始处。
                    // 即将索引 p = position() 处的字节复制到索引 0 处,将索引 p + 1 处的字节复制到索引 1 处,依此类推,直到将索引 limit() - 1 处的字节复制到索引
                    // n = limit() - 1 - p 处。然后将缓冲区的位置设置为 n+1,并将其界限设置为其容量。如果已定义了标记,则丢弃它。
                    //将缓冲区的位置设置为复制的字节数,而不是零,以便调用此方法后可以紧接着调用另一个相对 put 方法。
                    //从缓冲区写入数据之后调用此方法,以防写入不完整。例如,以下循环语句通过 buf 缓冲区将字节从一个信道复制到另一个信道:
                    byteBuffer.compact();
                }

                //执行操作的时候,移除避免下一次循环干扰
//                    原因是Selector不会自己从已选择键集中移除SelectionKey实例。必须在处理完通道时自己移除。下次该通道变成就绪时,Selector会再次将其放入已选择键集中
                keyIterable.remove();

            }

        }

    }
}

 

说实话,这个服务端当时花了好大的力气写完的,mmp,因为客户端进行可以不关闭通道直接kill,导致服务端并不知道对端已经离线,到时候服务端不断再进行空轮训,一旦进行read就抛出io异常!!

好了,服务和注册写完了,那么我们注册一把呗

 

@Test
    public void test1() throws Exception {
        TimeQueryServer timeQueryServer = new TimeQueryServer();

        // 构造zk客户端连接
        timeQueryServer.connectZK();

        // 注册服务器信息
        timeQueryServer.registerServerInfo("192.168.1.7", "8888");

        // 启动业务线程开始处理业务
        new TimeQueryService(Integer.parseInt("8888")).start();

        while(true) {
            Thread.sleep(200000);
//            System.out.println("..");
        }
    }

 

 不要在意那个null异常,那是因为我爸watch设置为null的原因

 

服务端写完了,我们再考虑写一波客户端

package cn.cutter.demo.hadoop.zookeeper;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;

import java.net.InetSocketAddress;
import java.net.SocketException;
import java.nio.ByteBuffer;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

/**
 * @ProjectName: cutter-point
 * @Package: cn.cutter.demo.hadoop.zookeeper
 * @ClassName: Consumer
 * @Author: xiaof
 * @Description: ${description}
 * @Date: 2019/4/3 14:11
 * @Version: 1.0
 */
public class Consumer {

    // 定义一个list用于存放最新的在线服务器列表
    private volatile ArrayList<String> onlineServers = new ArrayList<>();

    // 构造zk连接对象
    ZooKeeper zk = null;

    // 构造zk客户端连接
    public void connectZK() throws Exception {

        zk = new ZooKeeper("192.168.1.4:2181,192.168.1.4:2182,192.168.1.4:2183", 2000, new Watcher() {

            @Override
            public void process(WatchedEvent event) {
                if (event.getState() == Event.KeeperState.SyncConnected && event.getType() == Event.EventType.NodeChildrenChanged) {

                    try {
                        // 事件回调逻辑中,再次查询zk上的在线服务器节点即可,查询逻辑中又再次注册了子节点变化事件监听
                        getOnlineServers();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }

                }

            }
        });

    }

    // 查询在线服务器列表
    public void getOnlineServers() throws Exception {

        List<String> children = zk.getChildren("/servers", true);
        ArrayList<String> servers = new ArrayList<>();

        for (String child : children) {
            byte[] data = zk.getData("/servers/" + child, false, null);

            String serverInfo = new String(data);

            servers.add(serverInfo);
        }

        onlineServers = servers;
        System.out.println("查询了一次zk,当前在线的服务器有:" + servers);

    }

    public void sendRequest() throws Exception {
        Random random = new Random();
        while (true) {
            try {
                // 挑选一台当前在线的服务器
                int nextInt = random.nextInt(onlineServers.size());
                String server = onlineServers.get(nextInt);
                String hostname = server.split(":")[0];
                int port = Integer.parseInt(server.split(":")[1]);

                System.out.println("本次请求挑选的服务器为:" + server);

                //2.打开socket信道,设置成非阻塞模式
                SocketChannel socketChannel = SocketChannel.open();
                socketChannel.configureBlocking(false);

                //3.尝试建立连接,然后轮询,判定,连接是否完全建立
                int times = 0;
                if(!socketChannel.connect(new InetSocketAddress(hostname, port))) {
                    while(!socketChannel.finishConnect()) {
//                        System.out.println(times++ + ". ");
                    }
                }

                //4.创建相应的buffer缓冲
                ByteBuffer writeBuffer = ByteBuffer.wrap("test\n".getBytes());
                ByteBuffer readBuffer = ByteBuffer.allocate(1024);
                int totalBytesRcvd = 0;
                int bytesRcvd;

                //5.向socket信道发送数据,然后尝试读取数据
                socketChannel.write(writeBuffer);
                //读取数据
                if((bytesRcvd = socketChannel.read(readBuffer)) == -1) {
                    //这种非阻塞模式,如果读取不到数据是会返回0的,如果是-1该通道已到达流的末尾
                    throw new SocketException("连接关闭??");
                }

                //不停尝试获取数据,这是因为服务端数据反馈太慢了???
                while (bytesRcvd == 0) {
                    bytesRcvd = socketChannel.read(readBuffer);
                }

                //6.输出
                readBuffer.flip();
                byte reads[] = new byte[readBuffer.limit()];
                readBuffer.get(reads, 0, reads.length);
                System.out.println("收到信息:" + new String(reads));

                //7.关闭信道
                socketChannel.close();

                Thread.sleep(3000);
            } catch (Exception e) {
                e.printStackTrace();
            }

        }

    }

    public static void main(String[] args) throws Exception {

        Consumer consumer = new Consumer();
        // 构造zk连接对象
        consumer.connectZK();

        // 查询在线服务器列表
        consumer.getOnlineServers();

        // 处理业务(向一台服务器发送时间查询请求)
        consumer.sendRequest();

    }



}

启动客户端:

 

 

 

 为了体现zookeeper监控服务是否在线的操作,我们多起几个服务端,然后监控客户端的信息展示

 

 我们再起一个

 

 

 

 

 接下来我们kill掉8888端口的进程

 

 我们当前在线可以看到只有2个节点了,我们上zk看看,确实只有2个了

 

 

到这里zookeeper的上下线的判断已经完成,我最近再自学大数据的东西,想向大数据进军一波,欢迎大家一起探讨大数据的学习。

 

o
粉丝 0
博文 500
码字总数 0
作品 0
私信 提问
加载中
请先登录后再评论。
大数据技术之_06_Zookeeper学习_Zookeeper入门+Zookeeper安装+Zookeeper内部原理+Zookeeper实战(开发重点)+企业面试真题

第1章 Zookeeper入门1.1 概述1.2 特点1.3 数据结构1.4 应用场景1.5 下载地址第2章 Zookeeper安装2.1 本地模式安装部署2.2 配置参数解读第3章 Zookeeper内部原理3.1 选举机制(面试重点)3.2...

osc_x8jhvw7p
2019/02/23
2
0
zookeeper完整教程(尚硅谷)

第1章 Zookeeper入门 1.1 概述 Zookeeper是一个开源的分布式的,为分布式应用提供协调服务的Apache项目。 1.2 特点 1.3 数据结构 1.4 应用场景 提供的服务包括:统一命名服务、统一配置管理、...

osc_uspzomb8
2019/05/22
6
0
zookeeper实现动态感知服务器上下线

  在实际的生产环境中我们一般都是集群环境部署的,同一个程序我们会部署在相同的几台服务器中,这时我们可以通过负载均衡服务器去调度,但是我们并不能很快速的获知哪台服务器挂掉了,这时...

osc_7zyqbo9l
2019/02/27
1
0
大数据技术之Zookeeper

第1章 Zookeeper入门 1.1 概述 Zookeeper是一个开源的分布式的,为分布式应用提供协调服务的Apache项目。 1.2 特点 1.3 数据结构 1.4 应用场景 提供的服务包括:统一命名服务、统一配置管理、...

osc_3dxe2p2c
2019/04/02
4
0
尚硅谷大数据技术之Zookeeper第4章Zookeeper实战

4.1 分布式安装部署 0)集群规划 在hadoop102、hadoop103和hadoop104三个节点上部署Zookeeper。 1)解压安装 (1)解压zookeeper安装包到/opt/module/目录下 [atguigu@hadoop102 software]$ ...

osc_srt16prk
2018/07/24
6
0

没有更多内容

加载失败,请刷新页面

加载更多

聊聊rocketmq-client-go的TraceInterceptor

序 本文主要研究一下rocketmq-client-go的TraceInterceptor TraceInterceptor rocketmq-client-go-v2.0.0/producer/interceptor.go // WithTrace support rocketmq trace: https://github.c......

go4it
37分钟前
0
0
如何在Android文本视图周围添加边框? - How do I put a border around an Android textview?

问题: 是否可以在textview周围绘制边框? 解决方案: 参考一: https://stackoom.com/question/EfXR/如何在Android文本视图周围添加边框 参考二: https://oldbug.net/q/EfXR/How-do-I-put...

法国红酒甜
52分钟前
10
0
设计模式(4) 建造者模式

什么是建造者模式 经典建造者模式的优缺点 对建造者模式的扩展 什么是建造者模式 建造者模式将一个复杂的对象的构建与它的表示分离,使得同样的构建过程可以创建不同的表示。创建者模式隐藏了...

zhixin9001
今天
14
0
ArrayList源码分析 —— JDK8

ArrayList的特性 ArrayList内部使用数据作为存储结构,ArrayList可以理解为数组的扩展对象,封装了常用的和非常用的操作数组的方法。以及当数组长度不足以保存数组时,自动扩容数组,通常Arr...

XuePeng77
今天
56
0
__slots__的用法? - Usage of __slots__?

问题: Python中__slots__的目的是什么-尤其是关于何时以及何时不使用它的目的? 解决方案: 参考一: https://stackoom.com/question/1ymu/slots-的用法 参考二: https://oldbug.net/q/1ym...

富含淀粉
今天
17
0

没有更多内容

加载失败,请刷新页面

加载更多

返回顶部
顶部