每特教育学院蚂蚁课堂-3期分布式协调工具zookeeper实现选举策略与哨兵机制

Zookeeper一组通用服务——分布式同步（DistributedSynchronization）、命名服务（Naming分布式ZooKeeper集群（一个Leader，多个Follower），基于一定的策略来保证ZooKeeper3、Zookeeper所提供的服务涵盖：主从协调、服务器节点动态上下线、配置管理、分布式共享锁、统>一名称服务等管理(，)用户程序提交的数据（类似namenode中存放的metadata）；并为用户程序提供数据节点服务；ZookeeperZookeeperLeaderZookeeperZookeeper1、层次化 3、节点Znode可以包含数据节点（但是EPHEMERAL类型的节点不能有子节点）类型节点持久（persistent）（create-s/app1/test2“test2”客户端断开连接zk不删除 节点（默认是persistentcznode，znode数器，由父节点数据发布与订阅（配置中心生产者负载均衡：metaqbroker上的一个分区来发送消息，因此metaq在运行过程中，会把所有broker和对应的分区信息全部到ZK指定节点上，默认的策略是一个依次轮询的过程，生产者在通过ZK获取分区列表之后，会按照brokerId和partition的顺序排列组织成一个有序的分区列表，发送的区只会由一个消费者来消费。MetaQ在某个消费者故障或者重启等情况下，其他消费者会感知到这一变化（通过zookeeperwatch），然后重新进行负载均衡，保证所有的分区都有消费者进行消费。命名服务(Naming以是集群中的机器，提供的服务地址，对象等等——这些都可以统称他们为名（Name）ZK阿里巴巴开源的分布式服务框架Dubbo中使用ZooKeeper来作为其命名服务全局DubboDubbo服务提供者在启动的时候，向ZK上的指定节点/dubbo/${serviceName}/providers 写入自己的URL地址，这个操作就完成了服务的发布。 下的提供者URL地址，并向/dubbo/${serviceName}/consumers 下写入自己的URL地址。注意，所有向ZK上的地址都是临时节点，这样就能够保证服务提供者和消费者能够自动感应资源的变化。另外，Dubbo还有针对服务粒度的，方法是订阅 znode进行znode的变（包括znode本身内容及子节点的其中一个系统updateznode，那么另一个系统能够收到通知，并作出相应处理Master或者每个机器自己定时向系统汇报“我还活着”。这种做法可行，但是存在两个比较EPHEMERAL例如，系统在/clusterServers节点上一个Watcher，以后每动态加机器，那么/clusterServersEPHEMERAL/clusterServers/{hostname}.2.Master则是zookeeper中最为经典的应用场景了利用ZooKeeper的强一致性，能够保证在分布式高并况下节点创建的全局唯一性，即：同时有多个客户端请求创建/currentMaster另外，这种场景演化一下，就是动态Master。这就要用到EPHEMERAL_SEQUENTIAL类果的一种可能情况是这样：/currentMaster/{sessionId}-1,/currentMaster/{sessionId}-2,/currentMaster/{sessionId}-3Master，如果这个机器挂了，由于他Masterzkmasterhttpmaster。ZK上一些ROOT表的地址和HMaster的地址，HRegionServer也会把自己以临时节（Ephemeral）的方式到Zookeeper中，使得HMaster可以随时感知到各个HRegionServer的存活状态，同时，一旦HMaster出现问题，会重新出一个HMaster来HMaster分布式锁，这个主要得益于ZooKeeper为保证了数据的强一致性。锁服务可以分为两类，一个是保持独占，另一个是控制时序。zkznodecreateznode/distribute_lock局时序了。做法和上面基本类似，只是这里/distribuock已经预先存在，客户端在它Zookeeper安Zookeeperwindows安装Zookeeper.在官 zookeeper.我的在D:\machine新建data及 ZooKeeper的安装模式分为三种，分别为：单机模式（stand-alone）、集群模式和集群伪分布模式。ZooKeeperZooKeeperZooKeeper安装单击模式。至D:\machine\zookeeper-3.4.6\confzoo_sample.cfg并粘贴到当 下，命名zoo.cfg.Zookeeper集群环境搭建zk进行解压：tar-zxvfzookeeper-重命名：mvzookeeper-3.4.6viexportexportexportCLASSPATH=.:$JAVA_HOME/lib/dt.jar:$JAVA_HOME/lib/tools.jarexportPATH=$JAVA_HOME/bin:$ZOOKEEPER_HOME/bin:$PATHsourcecd/usr/local/zookeeper/confmvzoo_sample.cfgzoo.cfgconf:vizoo.cfgdataDir=/usr/local/zookeeper/data（zookeeperdata创建文件夹：mkdirmyid0：vimyid(内容为服务器标识：0)zookeeper进行 到hadoop01和还有/etc/profilehadoop01hadoop02myid1路径(vi路径：执行：zkServer.sh3状态：zkServer.shzkServer.shstatus##Thenumberofmillisecondsofeachtick#Thenumberofticksthattheinitial#synchronizationphasecantake#Thenumberofticksthatcanpass#sendingarequestandgettinganacknowledgement#thedirectorywherethesnapshotis#donotuse/tmpforstorage,/tmphereisjust#examplesakes.#theportatwhichtheclientswill#ZooKeeperserveranditsportno.#ZooKeeperensembleshouldknowabouteveryothermachineintheensemble#specifyserveridbycreating'myid'fileinthedataDir#usehostnameinsteadofIPaddressforconvenientmaintenance##Besuretoreadthemaintenancesectionof#administratorguidebeforeturningonautopurge.#ThenumberofsnapshotstoretainindataDir#autopurge.snapRetainCount=3#Purgetaskintervalin#Setto"0"todisableautopurgefeature<br>里（dataDir） dataLogDir：用于单独设置transactionlog的 ，transactionlog分离可以避免和普通log还有快照的竞争CLeaderZookeeper创建节点(znode方法1，InodeName2，javaHessian、Kryo3，Ids.OPEN_ACL_UNSAFE MavenZookeeperpublicpublicclassTest001//连接地privatestaticfinalStringADDRES=//session会privatestaticfinalintSESSION_OUTTIME=//信号量,阻塞程序执行,用户等待zookeeper连接成功,发送成功信号privatestaticfinalCountDownLatchcountDownLatch=newpublicstaticvoidmain(String[]args)throwsIOException,InterruptedException,KeeperExceptionZooKeeperzk=newZooKeeper(ADDRES,SESSION_OUTTIME,newWatcher()publicvoidprocess(WatchedEventevent)获取事件状KeeperStateKeeperStatekeeperState=EventTypeeventType=if(KeeperState.SyncConnected==keeperState)if(EventType.None==eventType)System.out.println("zk启动连接}}}Stringresult=zk.create("/itmayeidu_Lasting","Lasting".getBytes(),}}1.1.创建持久节点，并且允许任何服务器可以操Stringresult=zk.create("/itmayiedu_Lasting","Lasting".getBytes(),System.out.println("result:"+2.Stringresult=zk.create("/itmayiedu_temp","temp".getBytes(),Ids.OPEN_ACL_UNSAFE,System.out.println("result:"+在ZooKeeper中，接口类Watcher用于表示一个标准的事件处理器，其定义了事件通知相关的逻辑，包含KeeperState和EventType两个枚举类，分别代表了通知状态和事件类型，同时定义了事件的回调方法：process（WatchedEventevent）。Watcher接口7-3Watcher（-NodeD（-（-（-（-schema2：SASL表7-3中列举了ZooKeeper中最常见的几个通知状态和事件类型。process（）process方法是Watcher接口中的一个回调方法，当ZooKeeper向客户端发送一个Watcherpublicvoidprocess(WatchedEvent（EventType）和节（path），其数据结构如图7-5所示ZooKeeperWatchedEvent对象来封装服务端事件并传递给Watcher，从而方便回调方法process对服务端事件进行处都是对一个服务端事件的封装。不同的是，WatchedEventWatcherEventWatchedEventgetWrapperWatcherEventWatcherEventWatchedEventprocessprocess需要注意的一点是，无论是WatchedEvent还是WatcherEvent，其对ZooKeeper服务端事件的封装都是机及其简单的。举个例子来说，当/zk-book端会发送给客户端一个“ZNodeWatcherpublicpublicclassZkClientWatcherimplementsWatcher群连接地privatestaticfinalStringCONNECT_ADDRES=话超时时privatestaticfinalintSESSIONTIME=//信号量,让zk在连接之前等待,连接成功后才能往下走privatestaticfinalCountDownLatchcountDownLatchnewCountDownLatch(1);privatestaticStringLOG_MAIN="【main】";privateZooKeeperpublicvoidcreateConnection(StringconnectAddres,intsessionTimeOut){try{zk=newZooKeeper(connectAddres,sessionTimeOut,this);System.out.println(LOG_MAIN+"zk开始启动连接服务器 }catch(Exceptione){}}publicbooleancreatePath(Stringpath,Stringdata){try{this.exists(path,this.zk.create(pathdata.getBytesIds.OPEN_ACL_UNSAFECreateMode.PERSISTENT);System.out.println(LOG_MAIN+"节点创建成功,Path:"+path+",data:"+data);}catch(Exceptione){returnfalse;}return}判断指定节点是否*@param节点路publicStatexists(Stringpath,booleanneedWatch){try{returnthis.zk.exists(path,}catch(Exceptione){returnnull;}}publicbooleanupdateNode(Stringpath,Stringdata)throwsKeeperException,InterruptedException{exists(path,true);this.zk.setData(path,data.getBytes(),-1);returnfalse;}publicvoidprocess(WatchedEventwatchedEvent)取事件状KeeperStatekeeperState=取事件类EventTypeeventType=//zk路Stringpath=System.out.println("进入到process()keeperState:"+keeperStateeventType:"+eventTypepath:断是否建立连if(KeeperState.SyncConnected==keeperState){if(EventType.None==eventType){//如果建立建立成功,让后程序往下走System.out.println(LOG_MAINzk立连接成功!");}elseif(EventType.NodeCreated==eventType){System.out.println(LOG_MAIN事件通知,新增node节点path);}elseif hanged==eventType)System.out.println(LOG_MAIN+"事件通知,当前node节点"+path+"被修 }elseif(EventType.NodeDeleted==eventType)System.out.println(LOG_MAIN+"事件通知,当前node节点"+path+"被删 }} }publicstaticvoidmain(String[]args)throwsKeeperException,InterruptedException{ZkClientWatcherzkClientWatcher=newZkClientWatcher();zkClientWatcher.createConnection(CONNECT_ADDRES,SESSIONTIME); booleancreateResult=zkClientWatcher.createPath("/p15","pa-644064");}}}使用Zookeeper实现负载均衡原使用Zookeeper实现负载均衡原理，服务器端将启动的服务到，zk中心上，采用临时节点。客户端从zk节点上获取最MavenServerZkServerScoekt//##ServerScoekt//##ServerScoekt服务publicclassZkServerScoektimplementsRunnableprivateintport=publicstaticvoidmain(String[]args)throwsIOExceptionintport=ZkServerScoektserver=newThreadthread=new}publicZkServerScoekt(intport)this.port=}publicvoidrun()ServerSocketserverSocket=tryserverSocket=newSystem.out.println("Serverstartport:"+Socketsocket=while(true)socket=newThread(new}}catch(Exceptione)}finallytryif(serverSocket!=null)}}}catch(Exceptione2)}}}}publicpublicclassServerHandlerimplementsRunnableprivateSocketpublicServerHandler(Socketsocket)this.socket=}publicvoidrun(){BufferedReaderin=null;PrintWriterout=null;try{in=newBufferedReader(newInputStreamReader(this.socket.getInputStream()));out=newPrintWriter(this.socket.getOutputStream(),true);Stringbody=while(true)body=if(body==null)System.out.println("Receive:"+body); o,"+body);}}catch(Exceptione)if(in!=null)try}catch(IOExceptione1){}}if(out!=null){}if(this.socket!=null)try}catch(IOExceptione1){}this.socket=}}}}}publicpublicclassZkServerClientpublicstaticList<String>listServer=newpublicstaticvoidmain(String[]args) client=newBufferedReaderconsole=newBufferedReader(newwhile(true)Stringtryname=if("exit".equals(name))}}catch(IOExceptione)}}} 所有publicstaticvoidinitServer()}获取当前server信publicstaticStringgetServer()return}publicvoidsend(Stringname)Stringserver=String[]cfg=Socketsocket=BufferedReaderin=PrintWriterout=trysocket=newSocket(cfg[0],in=newBufferedReader(newout=newPrintWriter(socket.getOutputStream(),while(true)Stringresp=if(resp==elseif(resp.length()>0)System.out.println("Receive:"+}}}catch(Exceptione)}finallyif(out!=null)}if(in!=null)try}catch(IOExceptione)}}if(socket!=null)try}catch(IOExceptione)}}}}}publicclassZkServerScoektimplementsRunnableprivatestaticintport=publicstaticvoidmain(String[]args)throwsIOExceptionZkServerScoektserver=newThreadthread=new}publicZkServerScoekt(intport)this.port=}publicvoidregServer()//向 当前服务ZkClientclient=newZkClient(":2181",60000,Stringpath="/test/server"+ifclient.createEphemeral(path,":"+}publicvoidrun()ServerSocketserverSocket=tryserverSocket=newSystem.out.println("Serverstartport:"+Socketsocket=while(true)socket=newThread(new}}catch(Exceptione)}finallytryif(serverSocket!=null)}}catch(Exceptione2)}}}}publicpublicclassZkServerClientpublicstaticList<String>listServer=newpublicstaticStringparent=publicstaticvoidmain(String[]args)ZkServerClientclient=newBufferedReaderconsole=newBufferedReader(newwhile(true)Stringtryname=if("exit".equals(name))}clien }catch(IOExceptione)}}} 所有publicstaticvoidinitServer()//finalZkClientzkClient=newZkClient(":2181",6000,List<String>children=getChilds(zkClient, 事publicvoidhandleChildChange(StringparentPath,List<String>currentChilds)throwsExceptiongetChilds(zkClient,}}privatestaticvoidgetChilds(ZkClientzkClient,List<String>currentChilds)for(Stringp:currentChilds)StringpathValue=(String)zkClient.readData(parent+"/"+}serverCount=System.out.println("从 到信息:"+}请求次privatestaticintreqestCount=服务数privatestaticintserverCount=获取当前server信publicstaticStringgetServer()实现负载均StringserverName=listServer.get(reqestCount%return}publicvoidsend(Stringname)Stringserver=String[]cfg=Socketsocket=BufferedReaderin=PrintWriterout=trysocket=newSocket(cfg[0],in=newBufferedReader(newout=newPrintWriter(socket.getOutputStream(),while(true)Stringresp=if(resp==elseif(resp.length()>0)System.out.println("Receive:"+}}}catch(Exceptione)}finallyifif(out!=null)}if(in!=null)try}catch(IOExceptione)}}if(socket!=null)try}catch(IOExceptione)}}}}}使用Zookeepers实现分布式多线线程安数据问题，也就是线程安全问题。但是做读操作是不会发生数据问题。Lock锁机制，保证在多个线程共享同一个变量只能有一Java内存模型共享内存模型指的就是Ja内存模型简称JJMM决定一个线程对共享变量的写入时,能对另一个线程可见JM定义了线程和主内存之间的抽象关系：线程之间的共享变量在主内存（iner）中，每个线程都有一个私有的本地内存（loler），本地内存中了该线程以读写共享变量的副本本地内存是JM的一个抽象概念并不真实存在它涵盖了缓存写缓冲区寄存器以及其他的硬件和编译器优化。从上图来看，线程A与线程B之间如要通信的话，必须要经历下面2个步骤首先，线程A把本地内存A中更新过的共享变量刷新到主内存中去然后，线程B到主内存中去线程A之前已更新过的共享变量下面通过示意图来说明这两个步骤如上图所示，本地内存A和B有主内存享变量x的副本。假设初始时，这三个内存中的x值都为0。线程A在执行时，把更新后的x值（假设值为1）临时存放在自己的本地内存A中。当线程A和线程B需要通信时，线程A首先会把自己本地内存中修改后的x值刷新到主内存中，此时主内存中的x值变为了1。随后，线程B到主内存中去线程A更新后的x值，此时线程B的本地内存的x值也变为了1。存之间的交互，来为java程序员提供内存可见性保证。总结：Ja内存模型：ja内存模型简称jm，定义了一个线程对另一个线程可见。共享变量存放在主内存中，每个线程都有自己的本地内存，当多个线程同时一个数据的时候，可能本地内存没有及时刷新到主内存，所以就会发生线程安全问题。分布式锁解决办传统方式生成订单号( )Snowflake算法——////生成订单publicclassOrderNumGenerator//全局订单publicstaticintcount=publicStringgetNumber()try}catch(Exceptione)}SimpleDateFormatsimpt=newSimpleDateFormat("yyyy-MM-dd-HH-mm-returnsimpt.format(newDate())+"-"+}}////使用多线程模拟生成订单publicclassOrderServiceimplementsRunnableprivateOrderNumGeneratororderNumGenerator=newpublicvoidrun(){}publicvoidgetNumber()Stringnumber=orderNumGenerator.getNumber();System.Stringnumber=orderNumGenerator.getNumber();System.out.println(Thread.currentThread().getName生成订单ID:number);}publicstaticvoidmain(String[]args){for(inti=0;i<100;i++){newThread(new}}}synchronizedlocaSynchronized////使用多线程模拟生成订单publicclassOrderServiceimplementsRunnableprivateOrderNumGeneratororderNumGenerator=newpublicvoidrun()}publicvoidgetNumber()synchronizedsynchronized(this)Stringnumber=()+",+}}publicstaticvoidmain(String[]args)System.out.println("####生成唯一订单号OrderServiceorderService=newfor(inti=0;i<100;i++)new}}}LockpublicpublicclassOrderServiceimplementsRunnableprivateOrderNumGeneratororderNumGenerator=new使用lockprivatejava.util.concurrent.locks.Locklock=newpublicvoidrun()}publicpublicvoidgetNumber()try//synchronized(this)Stringnumber=()+",+//}catch(Exceptione)}finally}}publicstaticvoidmain(String[]args)OrderServiceorderServicenewOrderService();for(inti=0;i<100;i++)new}}}分布式场景下生成订单提前生成好，订单号，存放在redisredis使用redis使用zookeepertomcat应用，在执行定时任务时就会遇到同一任务可能执行务Zookeeperzookeeper创建临时序列节点来实现分布式锁，适用于顺序执行的程序，大体思路就是失，通过watch来节点的变化，从剩下的节点的找到最小的序列节点，获取分布式锁，MavenLockpublicpublicinterfaceLock//获取到锁的publicvoid publicvoid}}创建 Lock抽象////将重复代码写入子类中classLockimplementsLockzk连接地privatestaticfinalStringCONNECTSTRING=创建zk连protectedZkClientzkClient=newprotectedstaticfinalStringPATH=publicvoidgetLock()if(tryLock())System.out.println("##获取lock锁的资源}else等重新获取锁}}获取锁资booleanbooleanvoidpublicvoidunLock()if(zkClient!=null) 锁资源}}} ockpublicpublicclass ockextendsLockprivateCountDownLatchcountDownLatch=booleantryLock()tryreturn}catch(Exceptione)return}}voidwaitLock()IZkDataListenerizkDataListener=newIZkDataListener()publicvoidhandleDataDeleted(Stringpath)throwsException唤醒被等待的线if(countDownLatch!=null)}}publicvoid hange(Stringpath,Objectdata)throwsException} 事 hanges(P