基于ZooKeeper的分布式Session實現(xiàn)_已發(fā)布

1、PAGE29 / NUMPAGES29基于ZooKeeper的分布式Session實現(xiàn)1認識ZooKeeperZooKeeper “動物園管理員”。動物園里當然有好多的動物，游客可以根據(jù)動物園提供的向?qū)D到不同的場館觀賞各種類型的動物，而不是像走在原始叢林里，心驚膽顫的被動 物所觀賞。為了讓各種不同的動物呆在它們應(yīng)該呆的地方，而不是相互串門，或是相互廝殺，就需要動物園管理員按照動物的各種習(xí)性加以分類和管理，這樣我們才 能更加放心安全的觀賞動物。回到我們企業(yè)級應(yīng)用系統(tǒng)中，隨著信息化水平的不斷提高，我們的企業(yè)級系統(tǒng)變得越來越龐大臃腫，性能急劇下降，客戶抱怨頻頻。拆 分系統(tǒng)是目前我們可選擇的解決系統(tǒng)

2、可伸縮性和性能問題的唯一行之有效的方法。但是拆分系統(tǒng)同時也帶來了系統(tǒng)的復(fù)雜性各子系統(tǒng)不是孤立存在的，它們彼此 之間需要協(xié)作和交互，這就是我們常說的分布式系統(tǒng)。各個子系統(tǒng)就好比動物園里的動物，為了使各個子系統(tǒng)能正常為用戶提供統(tǒng)一的服務(wù)，必須需要一種機制來進 行協(xié)調(diào)這就是ZooKeeper動物園管理員。關(guān)于ZooKeeper更正式的介紹ZooKeeper是一個為分布式應(yīng)用程序提供高性能協(xié)調(diào)服務(wù)的工具集合。它可以應(yīng)用在一些需要提供統(tǒng)一協(xié)調(diào)服務(wù)的case中，例如命名、配置管理、同步和組服務(wù)等。而在我們的case中，它被作為一個協(xié)調(diào)分布式環(huán)境中各子系統(tǒng)之間共享狀態(tài)數(shù)據(jù)的基礎(chǔ)設(shè)施。2ZooKeeper之特

3、性ZooKeeper本質(zhì)上是一個分布式的小文件存儲系統(tǒng)。原本是Apache Hadoop的一個組件，現(xiàn)在被拆分為一個Hadoop的獨立子項目，在HBase（Hadoop的另外一個被拆分出來的子項目，用于分布式環(huán)境下的超大數(shù)據(jù)量的DBMS）中也用到了ZooKeeper集群。ZooKeeper有如下的特性：1）簡單ZooKeeper核心是一個精簡的文件系統(tǒng)，它提供了一些簡單的文件操作以與附加的功能，例如排序和通知。2）易表達ZooKeeper的數(shù)據(jù)結(jié)構(gòu)原型是一棵znode樹（類似Linux的文件系統(tǒng)），并且它們是一些已經(jīng)被構(gòu)建好的塊，可以用來構(gòu)建大型的協(xié)作數(shù)據(jù)結(jié)構(gòu)和協(xié)議。3）高可用性ZooKeep

4、er可以運行在一組服務(wù)器上，同時它們被設(shè)計成高可用性，為你的應(yīng)用程序避免單點故障。4）松耦合交互ZooKeeper提供的Watcher機制使得各客戶端與服務(wù)器的交互變得松耦合，每個客戶端無需知曉其他客戶端的存在，就可以和其他客戶端進行數(shù)據(jù)交互。5）豐富的APIZooKeeper為開發(fā)人員提供了一套豐富的API，減輕了開發(fā)人員編寫通用協(xié)議的負擔(dān)。這篇文章是關(guān)于如何在ZooKeeper上創(chuàng)建分布式Session系統(tǒng)，所以關(guān)于ZooKeeper的安裝、使用、管理等主題不在本文的討論圍，如果想了解ZooKeeper更加詳細的情況，請看另外一篇文章ZooKeeper實戰(zhàn)。3為什么使用ZooKeeper目

5、前有關(guān)于分布式Session的實現(xiàn)基本上都是基于memcached。memcached本質(zhì)上是一個存緩存系統(tǒng)。雖然memcached也可以是分布式集群環(huán)境的，但是對于一份數(shù)據(jù)來說，它總是存儲在某一臺memcached服務(wù)器上。如果發(fā)生網(wǎng)絡(luò)故障或是服務(wù)器當機，則存儲在這臺服務(wù)器上的所有數(shù)據(jù)都將不可訪問。由于數(shù)據(jù)是存儲在存中的，重啟服務(wù)器，將導(dǎo)致數(shù)據(jù)全部丟失。當然你可以自己實現(xiàn)一套機制，用來在分布式memcached之間進行數(shù)據(jù)的同步和持久化，但是實現(xiàn)這套機制談何容易！由上述ZooKeeper的特性可知，ZooKeeper是一個分布式小文件系統(tǒng)，并且被設(shè)計為高可用性。通過選舉算法和集群復(fù)制可以避免

6、單點故障，由于是文件系統(tǒng)，所以即使所有的ZooKeeper節(jié)點全部掛掉，數(shù)據(jù)也不會丟失，重啟服務(wù)器之后，數(shù)據(jù)即可恢復(fù)。另外ZooKeeper的節(jié)點更新是原子的，也就是說更新不是成功就是失敗。通過版本號，ZooKeeper實現(xiàn)了更新的樂觀鎖，當版本號不相符時，則表示待更新的節(jié)點已經(jīng)被其他客戶端提前更新了，而當前的整個更新操作將全部失敗。當然所有的一切ZooKeeper已經(jīng)為開發(fā)者提供了保障，我們需要做的只是調(diào)用API。有人會懷疑ZooKeeper的執(zhí)行能力，在ZooKeeper誕生的地方Y(jié)ahoo!給出了一組數(shù)據(jù)將打消你的懷疑。它的吞吐量標準已經(jīng)達到大約每秒10000基于寫操作的工作量。對于讀操

7、作的工作量來說，它的吞吐量標準還要高幾倍。4實現(xiàn)分布式Session所面臨的挑戰(zhàn)實現(xiàn)分布式session最大的挑戰(zhàn)莫過于如何實現(xiàn)session在分布式系統(tǒng)之間的共享。在分布式環(huán)境下，每個子系統(tǒng)都是跨網(wǎng)絡(luò)的獨立JVM，在這些JVM之間實現(xiàn)共享數(shù)據(jù)的方式無非就是TCP/IP通訊。無論是memcached，還是ZooKeeper，底層都是基于TCP/IP的。所以，我認為使用何種工具實現(xiàn)分布式Session都是可行的，沒有那種實現(xiàn)優(yōu)于另外一種實現(xiàn)，在不同的應(yīng)用場景，各有優(yōu)缺點。世間萬物，無十全十美，不要盲目的崇拜某種技術(shù)，唯有適合才是真理。1）Session ID的共享Session ID是一個實例化

8、Session對象的唯一標識，也是它在Web容器中可以被識別的唯一身份標簽。Jetty和Tomcat容器會通過一個Hash算法，得到一個唯一的ID字符串，然后賦值給某個實例化的Session，此時，這個Session就可以被放入Web容器的SessionManager中開始它短暫的一生。在Servlet中，我們可以通過 Session的getId()方法得到這個值，但是我們無法改變這個值。當Session走到它一生盡頭的時候，Web容器的SessionManager會根據(jù)這個ID將其“火化”。所以Session ID是非常重要的一個屬性，并且要保證它的唯一性。在單系統(tǒng)中，Session ID只

9、需要被自身的Web容器讀寫，但是在分布式環(huán)境中，多個Web容器需要共享同一個Session ID。因此，當某個子系統(tǒng)的Web容器產(chǎn)生一個新的ID時，它必須需要一種機制來通知其他子系統(tǒng)，并且告知新ID是什么。2）Session中數(shù)據(jù)的復(fù)制和共享Session ID的問題一樣，在分布式環(huán)境下，Session中的用戶數(shù)據(jù)也需要在各個子系統(tǒng)中共享。當用戶通過 Session的setAttribute()方法在Session中設(shè)置了一個用戶數(shù)據(jù)時，它只存在于當前與用戶交互的那個Web容器中，而對其他子系統(tǒng)的Web容器來說，這些數(shù)據(jù)是不可見的。當用戶在下一步跳轉(zhuǎn)到另外一個Web容器時，則又會創(chuàng)建一個新的S

10、ession對象，而此Session中并不包含上一步驟用戶設(shè)置的數(shù)據(jù)。其實Session在分布式系統(tǒng)之間的復(fù)制實現(xiàn)是簡單的，但是每次在Session數(shù)據(jù)發(fā)生變化時，都在子系統(tǒng)之間復(fù)制一次數(shù)據(jù)，會大大降低用戶的響應(yīng)速度。因此我們需要一種機制，即可以保證Session數(shù)據(jù)的一致性，又不會降低用戶操作的響應(yīng)度。3）Session的失效Session是有生命周期的，當Session的空閑時間(maxIdle屬性值)超出限制時，Session就失效了，這種設(shè)計主要是考慮到了Web容器的可靠性。當一個系統(tǒng)有上萬人使用時，就會產(chǎn)生上萬個Session對象，由于 的無狀態(tài)特性，服務(wù)器無法確切的知道用戶是否真的

11、離開了系統(tǒng)。因此如果沒有失效機制，所有被Session占據(jù)的存資源將永遠無法被釋放，直到系統(tǒng)崩潰為止。在分布式環(huán)境下，Session被簡單的創(chuàng)建，并且通過某種機制被復(fù)制到了其他系統(tǒng)中。你無法保證每個子系統(tǒng)的時鐘都是一致的，可能相差幾秒，甚至相差幾分鐘。當某個Web容器的Session失效時，可能其他的子系統(tǒng)中的Session并未失效，這時會產(chǎn)生一個有趣的現(xiàn)象，一個用戶在各個子系統(tǒng)之間跳轉(zhuǎn)時，有時會提示Session超時，而有時又能正常操作。因此我們需要一種機制，當某個系統(tǒng)的Session失效時，其他所有系統(tǒng)的與之相關(guān)聯(lián)的Session也要同步失效。4）類裝載問題在單系統(tǒng)環(huán)境下，所有類被裝載到“

12、同一個”ClassLoader中。我在同一個上打了引號，因為實際上并非是同一個ClassLoader，只是邏輯上我們認為是同一個。這里涉與到了JVM的類裝載機制，由于這個主題不是本文的討論重點，所以相關(guān)詳情可以參考相關(guān)的JVM文檔。因此即使是由memcached或是ZooKeeper返回的字節(jié)數(shù)組也可以正常的反序列化成相對應(yīng)的對象類型。但是在分布式環(huán)境下，問題就變得異常的復(fù)雜。我們通過一個例子來描述這個問題。用戶在某個子系統(tǒng)的Session中設(shè)置了一個User類型的對象，通過序列化，將User類型的對象轉(zhuǎn)換成字節(jié)數(shù)組，并通過網(wǎng)絡(luò)傳輸?shù)搅薽emcached或是ZooKeeper上。此時，用戶跳轉(zhuǎn)

13、到了另外一個子系統(tǒng)上，需要通過getAttribute方法從memcached或是ZooKeeper上得到先前設(shè)置的那個User類型的對象數(shù)據(jù)。但是問題出現(xiàn)了，在這個子系統(tǒng)的ClassLoader中并沒有裝載User類型。因此在做反序列化時出現(xiàn)了ClassNotFoundException異常。當然上面描述的4點挑戰(zhàn)只是在實現(xiàn)分布式Session過程中面臨的關(guān)鍵問題，并不是全部。其實在我實現(xiàn)分布式Session的整個過程中還遇到了其他的一些挑戰(zhàn)。比如，需要通過filter機制攔截 ServletRequest，以便覆蓋其getSession方法。但是在不同的Web容器中（例如Jetty或是To

14、mcat）對 ServletRequest的實現(xiàn)是不一樣的，雖然都是實現(xiàn)了 ServletRequest接口，但是各自又添加了一些特性在其中。例如，在Jetty容器中， Session的實現(xiàn)類是一個保護部類，無法從其繼承并覆蓋相關(guān)的方法，只能從其實現(xiàn)類的父類中繼承更加抽象的Session實現(xiàn)。這樣就會造成一個問題，我必須重新實現(xiàn)對Session整個生命周期管理的SessionManager接口。有人會說，那就放棄它的實現(xiàn)吧，我們自己實現(xiàn) Session接口。很不幸，那是不可能的。因為在Jetty的 ServletRequest實現(xiàn)類的一些方法中對Session的類型進行了強制轉(zhuǎn)換（轉(zhuǎn)換成它自定

15、義的 Session實現(xiàn)類），如果不從其繼承，則會出現(xiàn)ClassCastException異常。相比之下，Tomcat的對 ServletRequest和 Session接口的實現(xiàn)還是比較標準的。由此可見，實現(xiàn)分布式Session其實是和某種Web容器緊密耦合的。并不像網(wǎng)上有些人的輕描淡寫，僅僅覆蓋setAttribute和getAttribute方法是行不通的。5算法實現(xiàn)從上述的挑戰(zhàn)來看，要寫一個分布式應(yīng)用程序是困難的，主要原因是因為局部故障。由于數(shù)據(jù)需要通過網(wǎng)絡(luò)傳輸，而網(wǎng)絡(luò)是不穩(wěn)定的，所以如果網(wǎng)絡(luò)發(fā)生故障，則所有的數(shù)據(jù)通訊都將終止。ZooKeeper并不能解決網(wǎng)絡(luò)故障的發(fā)生，甚至它本身也是

16、基于網(wǎng)絡(luò)的分布式應(yīng)用程序。但是它為我們提供了一套工具集合，幫助我們建立安全處理局部故障的分布式應(yīng)用程序。接下來我們就開始描述如何實現(xiàn)基于ZooKeeper的分布式Session系統(tǒng)。1）基于ZooKeeper的分布式Session系統(tǒng)架構(gòu)為了實現(xiàn)高可用性，采用了ZooKeeper集群，ZooKeeper集 群是由一臺領(lǐng)導(dǎo)者服務(wù)器和若干臺跟隨者服務(wù)器構(gòu)成（總服務(wù)器數(shù)要奇數(shù)）。所有的讀操作由跟隨者提供，而寫操作由領(lǐng)導(dǎo)者提供，并且領(lǐng)導(dǎo)者還負責(zé)將寫入的數(shù)據(jù) 復(fù)制到集群中其他的跟隨者。當領(lǐng)導(dǎo)者服務(wù)器由于故障無法訪問時，剩下的所有跟隨者服務(wù)器就開始進行領(lǐng)導(dǎo)者的選舉。通過選舉算法，最終由一臺原本是跟隨者的

17、服務(wù)器升級為領(lǐng)導(dǎo)者。當然原來的領(lǐng)導(dǎo)者服務(wù)器一旦被恢復(fù)，它就只能作為跟隨者服務(wù)器，并在下一次選舉中爭奪領(lǐng)導(dǎo)者的位置。Web容器中的Session容器也將發(fā)生變化。它不再對用戶的Session進行本地管理，而是委托給ZooKeeper和我們自己實現(xiàn)的Session管理器。也就是說，ZooKeeper負責(zé)Session數(shù)據(jù)的存儲，而我們自己實現(xiàn)的Session管理器將負責(zé)Session生命周期的管理。最后是關(guān)于在分布式環(huán)境下共享Session ID的策略。我們還是通過客戶端的Cookie來實現(xiàn)，我們會自定義一個Cookie，并通過一定的算法在多個子系統(tǒng)之間進行共享。下面會對此進行詳細的描述。2）分布

18、式Session的數(shù)據(jù)模型Session數(shù)據(jù)的存儲是有一定格式的，下圖展示了一個Session ID為”1gyh0za3qmld7”的Session在ZooKeeper上的存儲結(jié)構(gòu)：“/SESSIONS”是一個組節(jié)點，用來在ZooKeeper上劃分不同功能組的定義。你可以把它理解為一個文件夾目錄。在這個目錄下可以存放0個或N個子節(jié)點，我們就把一個Session的實例作為一個節(jié)點，節(jié)點的名稱就是Session ID。在ZooKeeper中，每個節(jié)點本身也可以存放一個字節(jié)數(shù)組。因此，每個節(jié)點天然就是一個Key-Value鍵值對的數(shù)據(jù)結(jié)構(gòu)。我們將Session中的用戶數(shù)據(jù)（本質(zhì)上就是一個Map）設(shè)計

19、成多節(jié)點，節(jié)點名稱就是Session的key，而節(jié)點的數(shù)據(jù)就是Session的Value。采用這種設(shè)計主要是考慮到性能問題和ZooKeeper對節(jié)點大小的限制問題。當然，我們可以將Session中的用戶數(shù)據(jù)保存在一個Map中，然后將Map序列化之后存儲在對應(yīng)的Session節(jié)點中。但是大部分情況下，我們在讀取數(shù)據(jù)時并不需要整個Map，而是Map中的一個或幾個值。這樣就可以避免一個非常大的Map在網(wǎng)絡(luò)間傳來傳去。同理，在寫Session的時候，也可以最大限度的減少數(shù)據(jù)流量。另外由于ZooKeeper是一個小文件系統(tǒng)，為了性能，每個節(jié)點的大小為1MB。如果Session中的Map大于1MB，就不能

20、單節(jié)點的存儲了。當然，一個Key的數(shù)據(jù)量是很少會超過1MB的，如果真的超過1MB，你就應(yīng)該考慮一下，是否應(yīng)該將此數(shù)據(jù)保存在Session中。最后我們來關(guān)注一下Session節(jié)點中的數(shù)據(jù)SessionMetaData。它是一個Session實例的元數(shù)據(jù)，保存了一些與Session生命周期控制有關(guān)的數(shù)據(jù)。以下代碼就是SessionMetaData的實現(xiàn)：publicclassSessionMetaDataimplementsSerializable privatestaticfinallongserialVersionUID= -25L;privateStringid;/*session的創(chuàng)建時間

21、*/privateLongcreateTm;/*session的最大空閑時間*/privateLongmaxIdle;/*session的最后一次訪問時間*/privateLonglastAccessTm;/*是否可用*/privateBooleanvalidate=false;/*當前版本*/privateintversion= 0;/*構(gòu)造方法*/publicSessionMetaData() this.createTm= System.currentTimeMillis();this.lastAccessTm=this.createTm;this.validate=true;以下是N多g

22、etter和setter方法其中需要關(guān)注的屬性有：a)id屬性：Session實例的ID。b)maxIdle屬性：Session的最大空閑時間，默認情況下是30分鐘。c)lastAccessTm屬性：Session的最后一次訪問時間，每次調(diào)用Request.getSession方法時都會去更新這個值。用來計算當前Session是否超時。如果lastAccessTm+maxIdle小于System.currentTimeMillis()，就表示當前Session超時。d)validate屬性：表示當前Session是否可用，如果超時，則此屬性為false。e)version屬性：這個屬性是為了冗

23、余Znode的version值，用來實現(xiàn)樂觀鎖，對Session節(jié)點的元數(shù)據(jù)進行更新操作。這里有必要提一下一個老生常談的問題，就是所有存儲在節(jié)點上的對象必須是可序列化的，也就是必須實現(xiàn)Serializable接口，否則無法保存。這個問題在memcached和ZooKeeper上都存在的。3）實現(xiàn)過程實現(xiàn)分布式Session的第一步就是要定義一個filter，用來攔截 ServletRequest對象。以下代碼片段，展現(xiàn)了在Jetty容器下的filter實現(xiàn)。publicclassJettyDistributedSessionFilterextendsDistributedSessionFilt

24、er privateLoggerlog= Logger.getLogger(getClass();Overridepublicvoidinit(FilterConfig filterConfig)throwsServletException super.init(filterConfig);/實例化Jetty容器下的Session管理器sessionManager=newJettyDistributedSessionManager(conf);trysessionManager.start();/啟動初始化/創(chuàng)建組節(jié)點ZooKeeperHelper.createGroupNode();log.

25、debug(DistributedSessionFilter.init completed.);catch(Exception e) log.error(e);OverridepublicvoiddoFilter(ServletRequest request, ServletResponse response, FilterChain chain)throwsIOException,ServletException /Jetty容器的Request對象包裝器，用于重寫Session的相關(guān)操作JettyRequestWrapper req =newJettyRequestWrapper(requ

26、est,sessionManager);chain.doFilter(req, response);這個filter是繼承自DistributedSessionFilter的，這個父類主要是負責(zé)完成初始化參數(shù)設(shè)置等通用方法的實現(xiàn)，代碼如下所示：publicabstractclassDistributedSessionFilterimplementsFilter protectedLoggerlog= Logger.getLogger(getClass();/*參數(shù)配置*/protectedConfigurationconf;/*Session管理器*/protectedSessionManag

27、ersessionManager;/*初始化參數(shù)名稱*/publicstaticfinalStringSERVERS=servers;publicstaticfinalStringTIMEOUT=timeout;publicstaticfinalStringPOOLSIZE=poolsize;/*初始化*seejavax.servlet.Filter#init(javax.servlet.FilterConfig)*/Overridepublicvoidinit(FilterConfig filterConfig)throwsServletException conf=newConfigura

28、tion();String servers = filterConfig.getInitParameter(SERVERS);if(StringUtils.isNotBlank(servers) conf.setServers(servers);String timeout = filterConfig.getInitParameter(TIMEOUT);if(StringUtils.isNotBlank(timeout) tryconf.setTimeout(Long.valueOf(timeout);catch(NumberFormatException ex) log.error(tim

29、eout parse error+ timeout +.);String poolsize = filterConfig.getInitParameter(POOLSIZE);if(StringUtils.isNotBlank(poolsize) tryconf.setPoolSize(Integer.valueOf(poolsize);catch(NumberFormatException ex) log.error(poolsize parse error+ poolsize +.);/初始化ZooKeeper配置參數(shù)ZooKeeperHelper.initialize(conf);/*銷

30、毀*seejavax.servlet.Filter#destroy()*/Overridepublicvoiddestroy() if(sessionManager!=null) trysessionManager.stop();catch(Exception e) log.error(e);/銷毀ZooKeeperZooKeeperHelper.destroy();log.debug(DistributedSessionFilter.destroy completed.);在filter中需要關(guān)注的重點是doFilter方法。OverridepublicvoiddoFilter(Servle

31、tRequest request, ServletResponse response, FilterChain chain)throwsIOException,ServletException /Jetty容器的Request對象包裝器，用于重寫Session的相關(guān)操作JettyRequestWrapper req =newJettyRequestWrapper(request,sessionManager);chain.doFilter(req, response);這里實例化了一個包裝器（裝飾者模式）類，用來包裝Jetty容器的Request對象，并覆蓋其getSession方法。另外我們

32、還自己實現(xiàn)sessionManager接口，用來管理Session的生命周期。通過filter機制，我們就接管了Session的整個生命周期的管理權(quán)。接下來我們來看看，Request包裝器是如何重寫getSession方法，替換成使用ZooKeeper上的Session數(shù)據(jù)。關(guān)鍵代碼如下所示：Overridepublic Session getSession(booleancreate) /檢查Session管理器if(sessionManager=null& create) thrownewIllegalStateException(No SessionHandler or SessionM

33、anager);if(session!=null&sessionManager!=null) returnsession;session=null;/從客戶端cookie中查找Session IDString id =sessionManager.getRequestSessionId(request);log.debug(獲取客戶端的Session ID:+ id +);if(id !=null&sessionManager!=null) /如果存在，則先從管理器中取session=sessionManager.get Session(id,request);if(session=null&

34、 !create) returnnull;/否則實例化一個新的Session對象if(session=null&sessionManager!=null& create) session=sessionManager.new Session(request);returnsession;其實實現(xiàn)很簡單，大部分工作都委托給了sessionManager來處理。因此，還是讓我們來關(guān)注sessionManager的相關(guān)方法實現(xiàn)。A）獲取Session ID:OverridepublicStringgetRequestSessionId( ServletRequest request) returnC

35、ookieHelper.findSessionId(request);這個方法就是從客戶端的Cookies中查找我們的一個自定義的Cookie值，這個Cookie的名稱為：”DISTRIBUTED_SESSION_ID”(Web容器自己也在Cookie中寫了一個值，用來在不同的request中傳遞Session ID，這個Cookie的名稱叫“JSESSIONID”)。如果返回null，則表示客戶端從來都沒有創(chuàng)建過Session實例。B）如果返回的Cookie值不為null，則有3種可能性：其一，已經(jīng)實例化過一個Session對象并且可以正常使用；其二，雖然已經(jīng)實例化過了，但是可能此Sessi

36、on已經(jīng)超時失效；其三，分布式環(huán)境中的其他子系統(tǒng)已經(jīng)實例化過了，但是本系統(tǒng)中還未實例化過此Session對象。所以先要對已經(jīng)存在的Session ID進行處理。關(guān)鍵代碼如下：Overridepublic Session get Session(String id, ServletRequest request) /類型檢查if(!(requestinstanceofRequest) log.warn(不是Jetty容器下的Request對象);returnnull;/將 ServletRequest轉(zhuǎn)換成Jetty容器的Request類型Request req = (Request) requ

37、est;/ZooKeeper服務(wù)器上查找指定節(jié)點是否有效booleanvalid = ZooKeeperHelper.isValid(id);/如果為false，表示服務(wù)器上無該Session節(jié)點，需要重新創(chuàng)建(返回null)if(!valid) /刪除本地的副本sessions.remove(id);returnnull;else/更新Session節(jié)點的元數(shù)據(jù)ZooKeeperHelper.updateSessionMetaData(id); Session session =sessions.get(id);/如果存在，則直接返回if(session !=null) returnsess

38、ion;/否則創(chuàng)建指定ID的Session并返回(用于同步分布式環(huán)境中的其他機器上的Session本地副本)session =newJettyDistributedSession(AbstractSessionManager) req.getSessionManager(),System.currentTimeMillis(), id);sessions.put(id, session);returnsession;首先根據(jù)ID去ZooKeeper上驗證此Session是否有效，如果無效了，則直接返回null，表示此Session已經(jīng)超時不可用，同時需要刪除本地的“影子”Session對象（不

39、管存在與否）。如果該節(jié)點有效，則首先更新該Session節(jié)點的元數(shù)據(jù)（例如，最后一次訪問時間）。然后先到本地的Session容器中查找是否存在該ID的Session對象。本地Session容器中的Session對象并不用來保存用戶數(shù)據(jù)，也不進行生命周期管理，純粹為了在不同請求中進行傳遞。唯一有價值的就Session ID，因此，我喜歡把本地Session容器中的Session對象稱為“影子”Session，它只是ZooKeeper上真正Session的一個影子而已。如果Session節(jié)點沒有失效，但是本地Session容器并沒有指定ID的”影子”Session，則表示是第三種可能性，需要進行影

40、子Session的同步。正如代碼中所展示的，我們實例化一個指定ID的Session對象，并放入當前系統(tǒng)的Session容器中，這樣就完成了Session ID在分布式環(huán)境中的共享，以與Session對象在各子系統(tǒng)之間的同步。C）如果通過上面的方法返回的Session對象還是null，則真的需要實例化一個Session對象了，代碼如下所示：public Session new Session( ServletRequest request) /類型檢查if(!(requestinstanceofRequest) log.warn(不是Jetty容器下的Request對象);returnnull;

41、/將 ServletRequest轉(zhuǎn)換成Jetty容器的Request類型Request req = (Request) request;Session session =newJettyDistributedSession(AbstractSessionManager) req.getSessionManager(), request);add Session(session, request);String id = session.getId();/寫cookieCookie cookie = CookieHelper.writeSessionIdToCookie(id, req, re

42、q.getConnection().getResponse();if(cookie !=null) log.debug(Wrote sid to Cookie,name:+ cookie.getName() +,value:+ cookie.getValue() +);/在ZooKeeper服務(wù)器上創(chuàng)建session節(jié)點，節(jié)點名稱為Session ID/創(chuàng)建元數(shù)據(jù)SessionMetaData metadata =newSessionMetaData();metadata.setId(id);metadata.setMaxIdle(config.getTimeout() * 60 * 1000

43、);/轉(zhuǎn)換成毫秒ZooKeeperHelper.createSessionNode(metadata);returnsession;以上代碼會實例化一個Session對象，并將Session ID寫入客戶端Cookie中，最后實例化Session元數(shù)據(jù)，并在ZooKeeper上新建一個Session節(jié)點。通過上面步驟，我們就將Session的整個生命周期管理與ZooKeeper關(guān)聯(lián)起來了。接下來我們看看Session對象的幾個重要方法的重寫：publicsynchronizedObject getAttribute(String name) /獲取session IDString id = g

44、etId();if(StringUtils.isNotBlank(id) /返回Session節(jié)點下的數(shù)據(jù)returnZooKeeperHelper.getSessionData(id, name);returnnull;publicsynchronizedvoidremoveAttribute(String name) /獲取session IDString id = getId();if(StringUtils.isNotBlank(id) /刪除Session節(jié)點下的數(shù)據(jù)ZooKeeperHelper.removeSessionData(id, name);publicsynchroni

45、zedvoidsetAttribute(String name, Object value) /獲取session IDString id = getId();if(StringUtils.isNotBlank(id) /將數(shù)據(jù)添加到ZooKeeper服務(wù)器上ZooKeeperHelper.setSessionData(id, name, value);publicvoidinvalidate()throwsIllegalStateException /獲取session IDString id = getId();if(StringUtils.isNotBlank(id) /刪除Sessio

46、n節(jié)點ZooKeeperHelper.deleteSessionNode(id);這些方法中都是直接和ZooKeeper上對應(yīng)的Session進行數(shù)據(jù)交換。本來我是想在本地Session對象上創(chuàng)建一個ZooKeeper的緩沖，當用戶調(diào)用Session的讀方法時，先到本地緩沖中讀數(shù)據(jù)，讀不到再到ZooKeeper上去取，這樣可以減少網(wǎng)絡(luò)的通訊開銷。但在分布式環(huán)境下，這種策略所帶來的數(shù)據(jù)同步開銷更加的可觀。因為每次一個子系統(tǒng)的Session數(shù)據(jù)更新，都將觸發(fā)所有其他子系統(tǒng)與之關(guān)聯(lián)的Session數(shù)據(jù)同步操作，否則Session中數(shù)據(jù)的一致性將無法得到保障?？吹竭@里，大家可能已經(jīng)發(fā)覺了，所有與Zoo

47、Keeper交互的代碼都被封裝到ZooKeeperHelper類中，接下來就來看看這個類的實現(xiàn)。4）ZooKeeperHelper類實現(xiàn)publicclassZooKeeperHelper /*日志*/privatestaticLoggerlog=Logger.getLogger(ZooKeeperHelper.class);privatestaticStringhosts;privatestaticExecutorServicepool= Executors.newCachedThreadPool();privatestaticfinalStringGROUP_NAME=/SESSIONS;

48、/*初始化*/publicstaticvoidinitialize(Configuration config) hosts= config.getServers();/*銷毀*/publicstaticvoiddestroy() if(pool!=null) /關(guān)閉pool.shutdown();/*連接服務(wù)器*return*/publicstaticZooKeeper connect() ConnectionWatcher cw =newConnectionWatcher();ZooKeeper zk = cw.connection(hosts);returnzk;/*關(guān)閉一個會話*/pub

49、licstaticvoidclose(ZooKeeper zk) if(zk !=null) tryzk.close();catch(InterruptedException e) log.error(e);/*驗證指定ID的節(jié)點是否有效*paramid*return*/publicstaticbooleanisValid(String id) ZooKeeper zk =connect();if(zk !=null) tryreturnisValid(id, zk);finallyclose(zk);returnfalse;/*驗證指定ID的節(jié)點是否有效*paramid*paramzk*re

50、turn*/publicstaticbooleanisValid(String id, ZooKeeper zk) if(zk !=null) /獲取元數(shù)據(jù)SessionMetaData metadata =getSessionMetaData(id, zk);/如果不存在或是無效，則直接返回nullif(metadata =null) returnfalse;returnmetadata.getValidate();returnfalse;/*返回指定ID的Session元數(shù)據(jù)*paramid*return*/publicstaticSessionMetaData getSessionMet

51、aData(String id, ZooKeeper zk) if(zk !=null) String path =GROUP_NAME+/+ id;try/檢查節(jié)點是否存在Stat stat = zk.exists(path,false);/stat為null表示無此節(jié)點if(stat =null) returnnull;/獲取節(jié)點上的數(shù)據(jù)byte data = zk.getData(path,false,null);if(data !=null) /反序列化Object obj = SerializationUtils.deserialize(data);/轉(zhuǎn)換類型if(objinstan

52、ceofSessionMetaData) SessionMetaData metadata = (SessionMetaData) obj;/設(shè)置當前版本號metadata.setVersion(stat.getVersion();returnmetadata;catch(KeeperException e) log.error(e);catch(InterruptedException e) log.error(e);returnnull;/*更新Session節(jié)點的元數(shù)據(jù)*paramid Session ID*paramversion更新版本號*paramzk*/publicstaticv

53、oidupdateSessionMetaData(String id) ZooKeeper zk =connect();try/獲取元數(shù)據(jù)SessionMetaData metadata =getSessionMetaData(id, zk);if(metadata !=null) updateSessionMetaData(metadata, zk);finallyclose(zk);/*更新Session節(jié)點的元數(shù)據(jù)*paramid Session ID*paramversion更新版本號*paramzk*/publicstaticvoidupdateSessionMetaData(Ses

54、sionMetaData metadata, ZooKeeper zk) tryif(metadata !=null) String id = metadata.getId();Long now = System.currentTimeMillis();/當前時間/檢查是否過期Long timeout = metadata.getLastAccessTm() + metadata.getMaxIdle();/空閑時間/如果空閑時間小于當前時間，則表示Session超時if(timeout now) metadata.setValidate(false);log.debug(Session節(jié)點已

55、超時+ id +);/設(shè)置最后一次訪問時間metadata.setLastAccessTm(now);/更新節(jié)點數(shù)據(jù)String path =GROUP_NAME+/+ id;byte data = SerializationUtils.serialize(metadata);zk.setData(path, data, metadata.getVersion();log.debug(更新Session節(jié)點的元數(shù)據(jù)完成+ path +);catch(KeeperException e) log.error(e);catch(InterruptedException e) log.error(e

56、);/*返回ZooKeeper服務(wù)器上的Session節(jié)點的所有數(shù)據(jù)，并裝載為Map*paramid*return*/publicstaticMapgetSessionMap(String id) ZooKeeper zk =connect();if(zk !=null) String path =GROUP_NAME+/+ id;try/獲取元數(shù)據(jù)SessionMetaData metadata =getSessionMetaData(path, zk);/如果不存在或是無效，則直接返回nullif(metadata =null| !metadata.getValidate() return

57、null;/獲取所有子節(jié)點Listnodes = zk.getChildren(path,false);/存放數(shù)據(jù)MapsessionMap =newHashMap();for(String node : nodes) String dataPath = path +/+ node;Stat stat = zk.exists(dataPath,false);/節(jié)點存在if(stat !=null) /提取數(shù)據(jù)byte data = zk.getData(dataPath,false,null);if(data !=null) sessionMap.put(node, Serialization

58、Utils.deserialize(data);elsesessionMap.put(node,null);returnsessionMap;catch(KeeperException e) log.error(e);catch(InterruptedException e) log.error(e);finallyclose(zk);returnnull;/*創(chuàng)建一個組節(jié)點*/publicstaticvoidcreateGroupNode() ZooKeeper zk =connect();if(zk !=null) try/檢查節(jié)點是否存在Stat stat = zk.exists(GRO

59、UP_NAME,false);/stat為null表示無此節(jié)點，需要創(chuàng)建if(stat =null) /創(chuàng)建組件點String createPath = zk.create(GROUP_NAME,null, Ids.OPEN_ACL_UNSAFE,CreateMode.PERSISTENT);log.debug(創(chuàng)建節(jié)點完成:+ createPath +);elselog.debug(組節(jié)點已存在，無需創(chuàng)建+GROUP_NAME+);catch(KeeperException e) log.error(e);catch(InterruptedException e) log.error(e);

60、finallyclose(zk);/*創(chuàng)建指定Session ID的節(jié)點*paramsid Session ID*return*/publicstaticString createSessionNode(SessionMetaData metadata) if(metadata =null) returnnull;ZooKeeper zk =connect();/連接服務(wù)期if(zk !=null) String path =GROUP_NAME+/+ metadata.getId();try/檢查節(jié)點是否存在Stat stat = zk.exists(path,false);/stat為nu