- 浏览: 456497 次
- 性别:
- 来自: 杭州
-
文章分类
最新评论
-
goody9807:
goody9807 写道server.1和server.2启动 ...
深入浅出Zookeeper之五 Leader选举 -
goody9807:
server.1和server.2启动后,无法选出Leader ...
深入浅出Zookeeper之五 Leader选举 -
小黄牛:
分布式事务解决方案演示效果:http://www.iqiyi. ...
深入浅出Zookeeper之七分布式CREATE事务处理 -
chenghaitao111111:
兄弟syncedUpdateBrokersInfo这个代码和g ...
[metaq]Producer -
victory_gwb:
我看的3.4.6版本中takeSnapshot(); 这个方 ...
深入浅出Zookeeper之一Server启动
大名鼎鼎的Zookeeper是解决分布式问题的神器。小编最近简单阅读了代码,分享一下。有不对之处,还请大家指出。
整篇文章将分多个系列完成,因为涉及点比较多,很难在一片文章内搞定。关于zookeeper的使用场景,大家参考http://rdc.taobao.com/team/jm/archives/1232。api使用参考官网手http://zookeeper.apache.org/doc/trunk/。这里以最新的zookeeper3.4.5为例。
这个系列的第一篇来说说zookeeper server端的启动,以单机为例,分布式zookeeper将在后续专门分析。
单机版启动类ZooKeeperServerMain
protected void initializeAndRun(String[] args) throws ConfigException, IOException { try { ManagedUtil.registerLog4jMBeans(); } catch (JMException e) { LOG.warn("Unable to register log4j JMX control", e); } //解析配置文件zoo.cfg ServerConfig config = new ServerConfig(); if (args.length == 1) { config.parse(args[0]); } else { config.parse(args); } //启动 runFromConfig(config); }
具体解析:
public void parse(String path) throws ConfigException { QuorumPeerConfig config = new QuorumPeerConfig(); config.parse(path); // let qpconfig parse the file and then pull the stuff we are // interested in readFrom(config); }
启动
public void runFromConfig(ServerConfig config) throws IOException { LOG.info("Starting server"); try { // Note that this thread isn't going to be doing anything else, // so rather than spawning another thread, we will just call // run() in this thread. // create a file logger url from the command line args ZooKeeperServer zkServer = new ZooKeeperServer(); //2个文件,log和data文件 FileTxnSnapLog ftxn = new FileTxnSnapLog(new File(config.dataLogDir), new File(config.dataDir)); zkServer.setTxnLogFactory(ftxn); zkServer.setTickTime(config.tickTime); zkServer.setMinSessionTimeout(config.minSessionTimeout); zkServer.setMaxSessionTimeout(config.maxSessionTimeout); //连接工厂,默认NIOServerCnxnFactory cnxnFactory = ServerCnxnFactory.createFactory(); //初始化主线程,打开selector,并bind端口,打开NIO的ACCEPT通知 cnxnFactory.configure(config.getClientPortAddress(), config.getMaxClientCnxns()); //并生成最新的snapshot文件,启动IO主线程,从snapshot文件和log文件中恢复内存database结构和session结构 cnxnFactory.startup(zkServer); //启动线程等待之前启动的主线程结束 cnxnFactory.join(); if (zkServer.isRunning()) { zkServer.shutdown(); } } catch (InterruptedException e) { // warn, but generally this is ok LOG.warn("Server interrupted", e); } }
具体startup流程:
public void startup(ZooKeeperServer zks) throws IOException, InterruptedException { //启动IO主线程 start(); //从log和snapshot回复database和session,并重新生成一个最新的snapshot文件 zks.startdata(); //启动sessionTracker线程,初始化IO请求的处理链,并启动每个processor线程 zks.startup(); setZooKeeperServer(zks); }
具体恢复过程:
public void startdata() throws IOException, InterruptedException { //check to see if zkDb is not null if (zkDb == null) { //初始化database zkDb = new ZKDatabase(this.txnLogFactory); } if (!zkDb.isInitialized()) { loadData(); } }
DataTree用Map实现,key是节点名称,value是DataNode,DataNode从有parent指向父亲节点,有children指向所有孩子节点
public DataTree() { /* Rather than fight it, let root have an alias */ //'/','/zookeeper','/zookeeper/quota'3个系统节点初始化 nodes.put("", root); nodes.put(rootZookeeper, root); /** add the proc node and quota node */ root.addChild(procChildZookeeper); nodes.put(procZookeeper, procDataNode); procDataNode.addChild(quotaChildZookeeper); nodes.put(quotaZookeeper, quotaDataNode); }
具体恢复数据
public void loadData() throws IOException, InterruptedException { //执行恢复,并返回最新的事务ID setZxid(zkDb.loadDataBase()); // Clean up dead sessions //清理session LinkedList<Long> deadSessions = new LinkedList<Long>(); for (Long session : zkDb.getSessions()) { if (zkDb.getSessionWithTimeOuts().get(session) == null) { deadSessions.add(session); } } zkDb.setDataTreeInit(true); for (long session : deadSessions) { // XXX: Is lastProcessedZxid really the best thing to use? killSession(session, zkDb.getDataTreeLastProcessedZxid()); } //生成最新的snapshot文件 // Make a clean snapshot takeSnapshot(); }
load过程:
public long loadDataBase() throws IOException { //load过程中,发起分布式提议,对于单机版,先不考虑 PlayBackListener listener=new PlayBackListener(){ public void onTxnLoaded(TxnHeader hdr,Record txn){ Request r = new Request(null, 0, hdr.getCxid(),hdr.getType(), null, null); r.txn = txn; r.hdr = hdr; r.zxid = hdr.getZxid(); addCommittedProposal(r); } }; //load数据 long zxid = snapLog.restore(dataTree,sessionsWithTimeouts,listener); initialized = true;
restore过程:
public long restore(DataTree dt, Map<Long, Integer> sessions, PlayBackListener listener) throws IOException { //从FileSnap中恢复 snapLog.deserialize(dt, sessions); FileTxnLog txnLog = new FileTxnLog(dataDir); TxnIterator itr = txnLog.read(dt.lastProcessedZxid+1); long highestZxid = dt.lastProcessedZxid; TxnHeader hdr; //从snapshot中记录的最新的事务开始处理,将log中的事务merge到datatree中 while (true) { // iterator points to // the first valid txn when initialized hdr = itr.getHeader(); if (hdr == null) { //empty logs return dt.lastProcessedZxid; } if (hdr.getZxid() < highestZxid && highestZxid != 0) { LOG.error(highestZxid + "(higestZxid) > " + hdr.getZxid() + "(next log) for type " + hdr.getType()); } else { highestZxid = hdr.getZxid(); } try { processTransaction(hdr,dt,sessions, itr.getTxn()); } catch(KeeperException.NoNodeException e) { throw new IOException("Failed to process transaction type: " + hdr.getType() + " error: " + e.getMessage(), e); } listener.onTxnLoaded(hdr, itr.getTxn()); if (!itr.next()) break; } return highestZxid; }
FileSnap恢复过程:
public long deserialize(DataTree dt, Map<Long, Integer> sessions) throws IOException { // we run through 100 snapshots (not all of them) // if we cannot get it running within 100 snapshots // we should give up //找前100个snapshot文件,降序,最新的文件在最前面 List<File> snapList = findNValidSnapshots(100); if (snapList.size() == 0) { return -1L; } //从最新的文件开始恢复,如果反序列化ok而且checksum也ok,则恢复结束 File snap = null; boolean foundValid = false; for (int i = 0; i < snapList.size(); i++) { snap = snapList.get(i); InputStream snapIS = null; CheckedInputStream crcIn = null; try { LOG.info("Reading snapshot " + snap); snapIS = new BufferedInputStream(new FileInputStream(snap)); crcIn = new CheckedInputStream(snapIS, new Adler32()); InputArchive ia = BinaryInputArchive.getArchive(crcIn); deserialize(dt,sessions, ia); long checkSum = crcIn.getChecksum().getValue(); long val = ia.readLong("val"); if (val != checkSum) { throw new IOException("CRC corruption in snapshot : " + snap); } foundValid = true; break; } catch(IOException e) { LOG.warn("problem reading snap file " + snap, e); } finally { if (snapIS != null) snapIS.close(); if (crcIn != null) crcIn.close(); } } if (!foundValid) { throw new IOException("Not able to find valid snapshots in " + snapDir); } //snapshot文件名就记录着最新的zxid dt.lastProcessedZxid = Util.getZxidFromName(snap.getName(), "snapshot"); return dt.lastProcessedZxid; }
单个事务处理:
public void processTransaction(TxnHeader hdr,DataTree dt, Map<Long, Integer> sessions, Record txn) throws KeeperException.NoNodeException { ProcessTxnResult rc; switch (hdr.getType()) { //创建session case OpCode.createSession: sessions.put(hdr.getClientId(), ((CreateSessionTxn) txn).getTimeOut()); ...... // give dataTree a chance to sync its lastProcessedZxid rc = dt.processTxn(hdr, txn); break; case OpCode.closeSession: sessions.remove(hdr.getClientId()); if (LOG.isTraceEnabled()) { ZooTrace.logTraceMessage(LOG,ZooTrace.SESSION_TRACE_MASK, "playLog --- close session in log: 0x" + Long.toHexString(hdr.getClientId())); } rc = dt.processTxn(hdr, txn); break; default: rc = dt.processTxn(hdr, txn); } ...... }
DataTree处理单个事务
public ProcessTxnResult processTxn(TxnHeader header, Record txn) { ProcessTxnResult rc = new ProcessTxnResult(); try { rc.clientId = header.getClientId(); rc.cxid = header.getCxid(); rc.zxid = header.getZxid(); rc.type = header.getType(); rc.err = 0; rc.multiResult = null; switch (header.getType()) { case OpCode.create: CreateTxn createTxn = (CreateTxn) txn; rc.path = createTxn.getPath(); createNode( createTxn.getPath(), createTxn.getData(), createTxn.getAcl(), createTxn.getEphemeral() ? header.getClientId() : 0, createTxn.getParentCVersion(), header.getZxid(), header.getTime()); break; case OpCode.delete: DeleteTxn deleteTxn = (DeleteTxn) txn; rc.path = deleteTxn.getPath(); deleteNode(deleteTxn.getPath(), header.getZxid()); break; case OpCode.setData: SetDataTxn setDataTxn = (SetDataTxn) txn; rc.path = setDataTxn.getPath(); rc.stat = setData(setDataTxn.getPath(), setDataTxn .getData(), setDataTxn.getVersion(), header .getZxid(), header.getTime()); break; 》 ...... /* * A snapshot might be in progress while we are modifying the data * tree. If we set lastProcessedZxid prior to making corresponding * change to the tree, then the zxid associated with the snapshot * file will be ahead of its contents. Thus, while restoring from * the snapshot, the restore method will not apply the transaction * for zxid associated with the snapshot file, since the restore * method assumes that transaction to be present in the snapshot. * * To avoid this, we first apply the transaction and then modify * lastProcessedZxid. During restore, we correctly handle the * case where the snapshot contains data ahead of the zxid associated * with the file. */ //处理完事务后,再修改最新Zxid,如果是先修改Zxid再处理事务,修改完Zxid后,正好异步线程flush datatree //此时由于事务并没有被处理,导致snapshot中的zxid比content新,而restore的时候是从最新zxid+1开始恢复的,从而 //导致丢数据 if (rc.zxid > lastProcessedZxid) { lastProcessedZxid = rc.zxid; } ...... return rc; }
以上就完成了server的数据恢复过程,LSM的精华所在。
接下来server启动sessionTracker线程和请求处理链
protected void setupRequestProcessors() { RequestProcessor finalProcessor = new FinalRequestProcessor(this); RequestProcessor syncProcessor = new SyncRequestProcessor(this, finalProcessor); ((SyncRequestProcessor)syncProcessor).start(); firstProcessor = new PrepRequestProcessor(this, syncProcessor); ((PrepRequestProcessor)firstProcessor).start(); }
核心IO线程
public void run() { while (!ss.socket().isClosed()) { try { //select过程 selector.select(1000); Set<SelectionKey> selected; synchronized (this) { selected = selector.selectedKeys(); } ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>( selected); //打乱顺序 Collections.shuffle(selectedList); for (SelectionKey k : selectedList) { //新连接进来,accept之 if ((k.readyOps() & SelectionKey.OP_ACCEPT) != 0) { SocketChannel sc = ((ServerSocketChannel) k .channel()).accept(); InetAddress ia = sc.socket().getInetAddress(); int cnxncount = getClientCnxnCount(ia); //校验同个client连接数是否超过限制 if (maxClientCnxns > 0 && cnxncount >= maxClientCnxns){ LOG.warn("Too many connections from " + ia + " - max is " + maxClientCnxns ); sc.close(); } else { LOG.info("Accepted socket connection from " + sc.socket().getRemoteSocketAddress()); //异步模式 sc.configureBlocking(false); //监听read事件 SelectionKey sk = sc.register(selector, SelectionKey.OP_READ); //创建内部连接 NIOServerCnxn cnxn = createConnection(sc, sk); sk.attach(cnxn); //添加到连接表,方便后续统计 addCnxn(cnxn); } } //如果是read和write事件,则处理之 else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) { NIOServerCnxn c = (NIOServerCnxn) k.attachment(); c.doIO(k); } else { if (LOG.isDebugEnabled()) { LOG.debug("Unexpected ops in select " + k.readyOps()); } } } //准备下次IO selected.clear(); } catch (RuntimeException e) { LOG.warn("Ignoring unexpected runtime exception", e); } catch (Exception e) { LOG.warn("Ignoring exception", e); } } closeAll(); LOG.info("NIOServerCnxn factory exited run method"); }
具体io处理过程,将在后续结合实例来讲解。
至此server启动完成,就等待client去连接了。server启动核心功能就是从snapshot和log文件中恢复datatree,其核心就是zxid,典型的LSM应用。
评论
2 楼
victory_gwb
2016-02-25
我看的3.4.6版本中takeSnapshot(); 这个方法在SyncRequestProcessor中引用,没有在loadData()中呀
1 楼
迪伦少校
2015-10-26
如此好文竟然没人顶,赞一个!最近也在学习zookeeper,希望多交流!
相关推荐
本文旨在深入浅出地解析Zookeeper的核心概念和技术细节,帮助读者更好地理解和应用这一强大的工具。 #### 概述 Zookeeper是Apache基金会下的一个顶级项目,主要用于解决分布式环境中常见的协调问题,如命名服务、...
这份培训资料深入浅出地介绍了如何在Hadoop集群中使用HBase进行应用开发。 HBase基于Google的Bigtable设计理念,设计目标是提供高可靠性和高性能的数据存储。它以行键、列族、时间戳为索引,可以处理PB级别的数据,...
本教程将深入浅出地讲解如何在Spring项目中使用Seata进行分布式事务管理。 Seata,全称Simple Extensible Autonomous Transaction Architecture,是一个开源的分布式事务解决方案,致力于提供高性能和简单易用的...
在全球建筑行业不断追求节能与智能化发展的浪潮中,变风量(VAV)系统市场正展现出蓬勃的发展潜力。根据 QYResearch 报告出版商的深入调研统计,预计到 2031 年,全球变风量(VAV)系统市场销售额将飙升至 1241.3 亿元,在 2025 年至 2031 年期间,年复合增长率(CAGR)为 5.8%。这一令人瞩目的数据,不仅彰显了 VAV 系统在当今建筑领域的重要地位,更预示着其未来广阔的市场前景。 变风量系统的起源可追溯到 20 世纪 60 年代的美国。它犹如建筑空调系统中的 “智能管家”,能够敏锐地感知室内负荷或室内所需参数的变化,通过维持恒定的送风温度,自动、精准地调节空调系统的送风量,从而确保室内各项参数始终满足空调系统的严格要求。从系统构成来看,变风量系统主要由四个基本部分协同运作。变风量末端设备,包括 VAV 箱和室温控制器,如同系统的 “神经末梢”,负责接收室内环境变化的信号并做出初步响应;空气处理及输送设备则承担着对空气进行净化、加热、冷却等处理以及高效输送的重任;风管系统,涵盖新风、排风、送风、回风等管道,构建起了空气流通的 “高速公路”;而自动控制系统宛
资源内项目源码是来自个人的毕业设计,代码都测试ok,包含源码、数据集、可视化页面和部署说明,可产生核心指标曲线图、混淆矩阵、F1分数曲线、精确率-召回率曲线、验证集预测结果、标签分布图。都是运行成功后才上传资源,毕设答辩评审绝对信服的保底85分以上,放心下载使用,拿来就能用。包含源码、数据集、可视化页面和部署说明一站式服务,拿来就能用的绝对好资源!!! 项目备注 1、该资源内项目代码都经过测试运行成功,功能ok的情况下才上传的,请放心下载使用! 2、本项目适合计算机相关专业(如计科、人工智能、通信工程、自动化、电子信息等)的在校学生、老师或者企业员工下载学习,也适合小白学习进阶,当然也可作为毕设项目、课程设计、大作业、项目初期立项演示等。 3、如果基础还行,也可在此代码基础上进行修改,以实现其他功能,也可用于毕设、课设、作业等。 下载后请首先打开README.txt文件,仅供学习参考, 切勿用于商业用途。
内容概要:本文探讨了ChatGPT这种高级语音模式的人工智能聊天机器人与用户的互动对其情绪健康的影响。研究采用了两种互补的方法:大规模平台数据分析和随机对照试验(RCT)。平台数据部分通过对超过400万次对话进行隐私保护的大规模自动化分析以及对4000多名用户的调查,揭示了高频率使用者表现出更多的情感依赖和较低的社会交往意愿。RCT部分则通过近1000名参与者为期28天的研究,发现语音模型相较于文本模型能带来更好的情绪健康效果,但长时间使用可能导致负面后果。此外,初始情绪状态较差的用户在使用更具吸引力的语音模型时,情绪有所改善。 适合人群:对人机交互、情感计算和社会心理学感兴趣的科研人员和技术开发者。 使用场景及目标:本研究旨在为AI聊天机器人的设计提供指导,确保它们不仅能满足任务需求,还能促进用户的心理健康。同时,也为政策制定者提供了关于AI伦理使用的思考。 其他说明:研究强调了长期使用AI聊天机器人可能带来的复杂心理效应,特别是对于那些已经感到孤独或社交孤立的人来说,过度依赖可能会加剧这些问题。未来的研究应该更加关注这些极端情况下的用户体验。
Java 反射(Reflection)是一种强大的机制,允许程序在运行时检查和操作类的成员变量和方法。然而,传统的 `setAccessible(true)` 方式虽然便捷,但存在安全性问题,并且性能相对较低。在 Java 7 引入 `MethodHandle` 后,我们可以通过 `MethodHandles.Lookup.findVirtual()` 提供更优雅、高效的方式来访问对象属性。本文将对比这两种反射方式,并分析它们的优缺点。
loongdomShop.tar.gz
内容概要:本文探讨了不同交互模式(文本、中性语音、吸引人语音)和对话类型(开放式、非个人化、个人化)对聊天机器人使用者的心理社会效果(如孤独感、社交互动、情感依赖、不当使用)的影响。研究表明,在初期阶段,语音型聊天机器人比文本型更能缓解孤独感并减少情感依赖,但随着每日使用时间增加,这种优势逐渐消失,尤其是对于中性语音聊天机器人。此外,个人话题对话略微增加了孤独感,而非个人话题则导致更高的情感依赖。总体而言,高频率使用聊天机器人的用户表现出更多的孤独感、情感依赖和不当使用,同时减少了真实人际交往。研究还发现,某些个体特征(如依恋倾向、情绪回避)使用户更容易受到负面影响。 适合人群:心理学家、社会学家、人工智能研究人员以及关注心理健康和人机交互的专业人士。 使用场景及目标:①帮助理解不同类型聊天机器人对用户心理健康的潜在影响;②为设计更健康的人工智能系统提供指导;③制定政策和规范,确保聊天机器人的安全和有效使用。 其他说明:研究强调了进一步探索聊天机器人管理情感内容而不引发依赖或替代人际关系的重要性,呼吁更多跨学科的研究来评估长期影响。
MP4575GF-Z MP4575 TSSOP-20 降压型可调DC-DC电源芯片
界面设计_SwiftUI_习惯养成_项目管理_1742850611.zip
免安装版的logic软件包。支持波形实时查看。内含驱动文件。
1. **系统名称**:学生毕业离校系统 2. **技术栈**:Java技术、MySQL数据库、Spring Boot框架、B/S架构、Tomcat服务器、Eclipse开发环境 3. **系统功能**: - **管理员功能**:首页、个人中心、学生管理、教师管理、离校信息管理、费用结算管理、论文审核管理、管理员管理、留言板管理、系统管理。 - **学生功能**:首页、个人中心、费用结算管理、论文审核管理、我的收藏管理。 - **教师功能**:首页、个人中心、学生管理、离校信息管理、费用结算管理、论文审核管理。
配套文章:https://blog.csdn.net/gust2013/article/details/139608432
蓝凌OA系统V15.0管理员手册
资源内项目源码是来自个人的毕业设计,代码都测试ok,包含源码、数据集、可视化页面和部署说明,可产生核心指标曲线图、混淆矩阵、F1分数曲线、精确率-召回率曲线、验证集预测结果、标签分布图。都是运行成功后才上传资源,毕设答辩评审绝对信服的保底85分以上,放心下载使用,拿来就能用。包含源码、数据集、可视化页面和部署说明一站式服务,拿来就能用的绝对好资源!!! 项目备注 1、该资源内项目代码都经过测试运行成功,功能ok的情况下才上传的,请放心下载使用! 2、本项目适合计算机相关专业(如计科、人工智能、通信工程、自动化、电子信息等)的在校学生、老师或者企业员工下载学习,也适合小白学习进阶,当然也可作为毕设项目、课程设计、大作业、项目初期立项演示等。 3、如果基础还行,也可在此代码基础上进行修改,以实现其他功能,也可用于毕设、课设、作业等。 下载后请首先打开README.txt文件,仅供学习参考, 切勿用于商业用途。
mips-gcc520-glibc222编译工具链.zip
app开发
内容概要:本文档详细介绍了Swift编程语言的基础知识,涵盖语言特点、基础语法、集合类型、控制流、函数定义、面向对象编程、可选类型、错误处理、协议与扩展以及内存管理等方面的内容。此外还简要提及了Swift与UIKit/SwiftUI的关系,并提供了进一步学习的资源推荐。通过这份文档,读者可以全面了解Swift的基本概念及其在iOS/macOS/watchOS/tvOS平台的应用开发中的使用方法。 适合人群:初学者或者希望从其他编程语言转向Swift的开发者。 使用场景及目标:帮助读者快速上手Swift编程,掌握其基本语法和特性,能够独立完成简单的程序编写任务,为进一步学习高级主题如并发编程、图形界面设计打下坚实的基础。 阅读建议:由于Swift是一门现代化的语言,拥有许多独特的特性和最佳实践方式,在学习过程中应当多加练习并尝试理解背后的原理。同时利用提供的官方文档和其他辅助材料加深印象。
资源内项目源码是来自个人的毕业设计,代码都测试ok,包含源码、数据集、可视化页面和部署说明,可产生核心指标曲线图、混淆矩阵、F1分数曲线、精确率-召回率曲线、验证集预测结果、标签分布图。都是运行成功后才上传资源,毕设答辩评审绝对信服的保底85分以上,放心下载使用,拿来就能用。包含源码、数据集、可视化页面和部署说明一站式服务,拿来就能用的绝对好资源!!! 项目备注 1、该资源内项目代码都经过测试运行成功,功能ok的情况下才上传的,请放心下载使用! 2、本项目适合计算机相关专业(如计科、人工智能、通信工程、自动化、电子信息等)的在校学生、老师或者企业员工下载学习,也适合小白学习进阶,当然也可作为毕设项目、课程设计、大作业、项目初期立项演示等。 3、如果基础还行,也可在此代码基础上进行修改,以实现其他功能,也可用于毕设、课设、作业等。 下载后请首先打开README.txt文件,仅供学习参考, 切勿用于商业用途。