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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

15、義的 Session實(shí)現(xiàn)類），如果不從其繼承，則會(huì)出現(xiàn)ClassCastException異常。相比之下，Tomcat的對(duì) ServletRequest和 Session接口的實(shí)現(xiàn)還是比較標(biāo)準(zhǔn)的。由此可見(jiàn)，實(shí)現(xiàn)分布式Session其實(shí)是和某種Web容器緊密耦合的。并不像網(wǎng)上有些人的輕描淡寫(xiě)，僅僅覆蓋setAttribute和getAttribute方法是行不通的。5算法實(shí)現(xiàn)從上述的挑戰(zhàn)來(lái)看，要寫(xiě)一個(gè)分布式應(yīng)用程序是困難的，主要原因是因?yàn)榫植抗收?。由于?shù)據(jù)需要通過(guò)網(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)用程序。接下來(lái)我們就開(kāi)始描述如何實(shí)現(xiàn)基于ZooKeeper的分布式Session系統(tǒng)。1）基于ZooKeeper的分布式Session系統(tǒng)架構(gòu)為了實(shí)現(xiàn)高可用性，采用了ZooKeeper集群，ZooKeeper集 群是由一臺(tái)領(lǐng)導(dǎo)者服務(wù)器和若干臺(tái)跟隨者服務(wù)器構(gòu)成（總服務(wù)器數(shù)要奇數(shù)）。所有的讀操作由跟隨者提供，而寫(xiě)操作由領(lǐng)導(dǎo)者提供，并且領(lǐng)導(dǎo)者還負(fù)責(zé)將寫(xiě)入的數(shù)據(jù) 復(fù)制到集群中其他的跟隨者。當(dāng)領(lǐng)導(dǎo)者服務(wù)器由于故障無(wú)法訪問(wèn)時(shí)，剩下的所有跟隨者服務(wù)器就開(kāi)始進(jìn)行領(lǐng)導(dǎo)者的選舉。通過(guò)選舉算法，最終由一臺(tái)原本是跟隨者的

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

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

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

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

21、*/privateLongcreateTm;/*session的最大空閑時(shí)間*/privateLongmaxIdle;/*session的最后一次訪問(wèn)時(shí)間*/privateLonglastAccessTm;/*是否可用*/privateBooleanvalidate=false;/*當(dāng)前版本*/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實(shí)例的ID。b)maxIdle屬性：Session的最大空閑時(shí)間，默認(rèn)情況下是30分鐘。c)lastAccessTm屬性：Session的最后一次訪問(wèn)時(shí)間，每次調(diào)用Request.getSession方法時(shí)都會(huì)去更新這個(gè)值。用來(lái)計(jì)算當(dāng)前Session是否超時(shí)。如果lastAccessTm+maxIdle小于System.currentTimeMillis()，就表示當(dāng)前Session超時(shí)。d)validate屬性：表示當(dāng)前Session是否可用，如果超時(shí)，則此屬性為false。e)version屬性：這個(gè)屬性是為了冗

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

24、er privateLoggerlog= Logger.getLogger(getClass();Overridepublicvoidinit(FilterConfig filterConfig)throwsServletException super.init(filterConfig);/實(shí)例化Jetty容器下的Session管理器sessionManager=newJettyDistributedSessionManager(conf);trysessionManager.start();/啟動(dòng)初始化/創(chuàng)建組節(jié)點(diǎn)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對(duì)象包裝器，用于重寫(xiě)Session的相關(guān)操作JettyRequestWrapper req =newJettyRequestWrapper(requ

26、est,sessionManager);chain.doFilter(req, response);這個(gè)filter是繼承自DistributedSessionFilter的，這個(gè)父類主要是負(fù)責(zé)完成初始化參數(shù)設(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)注的重點(diǎn)是doFilter方法。OverridepublicvoiddoFilter(Servle

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

32、還自己實(shí)現(xiàn)sessionManager接口，用來(lái)管理Session的生命周期。通過(guò)filter機(jī)制，我們就接管了Session的整個(gè)生命周期的管理權(quán)。接下來(lái)我們來(lái)看看，Request包裝器是如何重寫(xiě)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;/否則實(shí)例化一個(gè)新的Session對(duì)象if(session=null&sessionManager!=null& create) session=sessionManager.new Session(request);returnsession;其實(shí)實(shí)現(xiàn)很簡(jiǎn)單，大部分工作都委托給了sessionManager來(lái)處理。因此，還是讓我們來(lái)關(guān)注sessionManager的相關(guān)方法實(shí)現(xiàn)。A）獲取Session ID:OverridepublicStringgetRequestSessionId( ServletRequest request) returnC

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

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

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

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

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

40、子Session的同步。正如代碼中所展示的，我們實(shí)例化一個(gè)指定ID的Session對(duì)象，并放入當(dāng)前系統(tǒng)的Session容器中，這樣就完成了Session ID在分布式環(huán)境中的共享，以與Session對(duì)象在各子系統(tǒng)之間的同步。C）如果通過(guò)上面的方法返回的Session對(duì)象還是null，則真的需要實(shí)例化一個(gè)Session對(duì)象了，代碼如下所示：public Session new Session( ServletRequest request) /類型檢查if(!(requestinstanceofRequest) log.warn(不是Jetty容器下的Request對(duì)象);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();/寫(xiě)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é)點(diǎn)，節(jié)點(diǎn)名稱為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;以上代碼會(huì)實(shí)例化一個(gè)Session對(duì)象，并將Session ID寫(xiě)入客戶端Cookie中，最后實(shí)例化Session元數(shù)據(jù)，并在ZooKeeper上新建一個(gè)Session節(jié)點(diǎn)。通過(guò)上面步驟，我們就將Session的整個(gè)生命周期管理與ZooKeeper關(guān)聯(lián)起來(lái)了。接下來(lái)我們看看Session對(duì)象的幾個(gè)重要方法的重寫(xiě)：publicsynchronizedObject getAttribute(String name) /獲取session IDString id = g

44、etId();if(StringUtils.isNotBlank(id) /返回Session節(jié)點(diǎn)下的數(shù)據(jù)returnZooKeeperHelper.getSessionData(id, name);returnnull;publicsynchronizedvoidremoveAttribute(String name) /獲取session IDString id = getId();if(StringUtils.isNotBlank(id) /刪除Session節(jié)點(diǎn)下的數(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é)點(diǎn)ZooKeeperHelper.deleteSessionNode(id);這些方法中都是直接和ZooKeeper上對(duì)應(yīng)的Session進(jìn)行數(shù)據(jù)交換。本來(lái)我是想在本地Session對(duì)象上創(chuàng)建一個(gè)ZooKeeper的緩沖，當(dāng)用戶調(diào)用Session的讀方法時(shí)，先到本地緩沖中讀數(shù)據(jù)，讀不到再到ZooKeeper上去取，這樣可以減少網(wǎng)絡(luò)的通訊開(kāi)銷。但在分布式環(huán)境下，這種策略所帶來(lái)的數(shù)據(jù)同步開(kāi)銷更加的可觀。因?yàn)槊看我粋€(gè)子系統(tǒng)的Session數(shù)據(jù)更新，都將觸發(fā)所有其他子系統(tǒng)與之關(guān)聯(lián)的Session數(shù)據(jù)同步操作，否則Session中數(shù)據(jù)的一致性將無(wú)法得到保障?？吹竭@里，大家可能已經(jīng)發(fā)覺(jué)了，所有與Zoo

47、Keeper交互的代碼都被封裝到ZooKeeperHelper類中，接下來(lái)就來(lái)看看這個(gè)類的實(shí)現(xiàn)。4）ZooKeeperHelper類實(shí)現(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)閉一個(gè)會(huì)話*/pub

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

50、turn*/publicstaticbooleanisValid(String id, ZooKeeper zk) if(zk !=null) /獲取元數(shù)據(jù)SessionMetaData metadata =getSessionMetaData(id, zk);/如果不存在或是無(wú)效，則直接返回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é)點(diǎn)是否存在Stat stat = zk.exists(path,false);/stat為null表示無(wú)此節(jié)點(diǎn)if(stat =null) returnnull;/獲取節(jié)點(diǎn)上的數(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è)置當(dāng)前版本號(hào)metadata.setVersion(stat.getVersion();returnmetadata;catch(KeeperException e) log.error(e);catch(InterruptedException e) log.error(e);returnnull;/*更新Session節(jié)點(diǎn)的元數(shù)據(jù)*paramid Session ID*paramversion更新版本號(hào)*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é)點(diǎn)的元數(shù)據(jù)*paramid Session ID*paramversion更新版本號(hào)*paramzk*/publicstaticvoidupdateSessionMetaData(Ses

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

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

56、);/*返回ZooKeeper服務(wù)器上的Session節(jié)點(diǎn)的所有數(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);/如果不存在或是無(wú)效，則直接返回nullif(metadata =null| !metadata.getValidate() return

57、null;/獲取所有子節(jié)點(diǎn)Listnodes = zk.getChildren(path,false);/存放數(shù)據(jù)MapsessionMap =newHashMap();for(String node : nodes) String dataPath = path +/+ node;Stat stat = zk.exists(dataPath,false);/節(jié)點(diǎn)存在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)建一個(gè)組節(jié)點(diǎn)*/publicstaticvoidcreateGroupNode() ZooKeeper zk =connect();if(zk !=null) try/檢查節(jié)點(diǎn)是否存在Stat stat = zk.exists(GRO

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

60、finallyclose(zk);/*創(chuàng)建指定Session ID的節(jié)點(diǎn)*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é)點(diǎn)是否存在Stat stat = zk.exists(path,false);/stat為nu