一、分布式session
session 是啥？瀏覽器有個 cookie，在一段時間內這個 cookie 都存在，然后每次發請求過來都帶上一個特殊的 jsessionid cookie，就根據這個東西，在服務端可以維護一個對應的 session 域，里面可以放點數據。
一般的話只要你沒關掉瀏覽器，cookie 還在，那么對應的那個 session 就在，但是如果 cookie 沒了，session 也就沒了。常見於什么購物車之類的東西，還有登錄狀態保存之類的。
這個不多說了，懂 Java 的都該知道這個。
單塊系統的時候這么玩兒 session 沒問題，但是你要是分布式系統呢，那么多的服務，session 狀態在哪兒維護啊？
（1）完全不用 session
使用 JWT Token 儲存用戶身份，然后再從數據庫或者 cache 中獲取其他的信息。這樣無論請求分配到哪個服務器都無所謂
（2）tomcat + redis
這個其實還挺方便的，就是使用 session 的代碼，跟以前一樣，還是基於 tomcat 原生的 session 支持即可，然后就是用一個叫做 Tomcat RedisSessionManager 的東西，讓所有我們部署的 tomcat 都將 session 數據存儲到 redis 即可。
在 tomcat 的配置文件中配置：
<Valve className="com.orangefunction.tomcat.redissessions.RedisSessionHandlerValve" /> <Manager className="com.orangefunction.tomcat.redissessions.RedisSessionManager" host="{redis.host}" port="{redis.port}" database="{redis.dbnum}" maxInactiveInterval="60"/>
然后指定 redis 的 host 和 port 就 ok 了。
<Valve className="com.orangefunction.tomcat.redissessions.RedisSessionHandlerValve" /> <Manager className="com.orangefunction.tomcat.redissessions.RedisSessionManager" sentinelMaster="mymaster" sentinels="<sentinel1-ip>:26379,<sentinel2-ip>:26379,<sentinel3-ip>:26379" maxInactiveInterval="60"/>
還可以用上面這種方式基於 redis 哨兵支持的 redis 高可用集群來保存 session 數據，都是 ok 的。
（3）spring session + redis
上面所說的第二種方式會與 tomcat 容器重耦合，如果我要將 web 容器遷移成 jetty，難道還要重新把 jetty 都配置一遍？
因為上面那種 tomcat + redis 的方式好用，但是會嚴重依賴於web容器，不好將代碼移植到其他 web 容器上去，尤其是你要是換了技術棧咋整？比如換成了 spring cloud 或者是 spring boot 之類的呢？
所以現在比較好的還是基於 Java 一站式解決方案，也就是 spring。人家 spring 基本上承包了大部分我們需要使用的框架，spirng cloud 做微服務，spring boot 做腳手架，所以用 sping session 是一個很好的選擇。
在 pom.xml 中配置：
<dependency> <groupId>org.springframework.session</groupId> <artifactId>spring-session-data-redis</artifactId> <version>1.2.1.RELEASE</version> </dependency> <dependency> <groupId>redis.clients</groupId> <artifactId>jedis</artifactId> <version>2.8.1</version> </dependency>
在 spring 配置文件中配置：
<bean id="redisHttpSessionConfiguration"
class="org.springframework.session.data.redis.config.annotation.web.http.RedisHttpSessionConfiguration">
<property name="maxInactiveIntervalInSeconds" value="600"/>
</bean>
<bean id="jedisPoolConfig" class="redis.clients.jedis.JedisPoolConfig">
<property name="maxTotal" value="100" />
<property name="maxIdle" value="10" />
</bean>
<bean id="jedisConnectionFactory"
class="org.springframework.data.redis.connection.jedis.JedisConnectionFactory" destroy-method="destroy">
<property name="hostName" value="${redis_hostname}"/>
<property name="port" value="${redis_port}"/>
<property name="password" value="${redis_pwd}" />
<property name="timeout" value="3000"/>
<property name="usePool" value="true"/>
<property name="poolConfig" ref="jedisPoolConfig"/>
</bean>
在 web.xml 中配置：
<filter>
<filter-name>springSessionRepositoryFilter</filter-name>
<filter-class>org.springframework.web.filter.DelegatingFilterProxy</filter-class>
</filter>
<filter-mapping>
<filter-name>springSessionRepositoryFilter</filter-name>
<url-pattern>/*</url-pattern>
</filter-mapping>
示例代碼：
@RestController @RequestMapping("/test") public class TestController { @RequestMapping("/putIntoSession") public String putIntoSession(HttpServletRequest request, String username) { request.getSession().setAttribute("name", "leo"); return "ok"; } @RequestMapping("/getFromSession") public String getFromSession(HttpServletRequest request, Model model){ String name = request.getSession().getAttribute("name"); return name; } }
上面的代碼就是 ok 的，給 sping session 配置基於 redis 來存儲 session 數據，然后配置了一個 spring session 的過濾器，這樣的話，session 相關操作都會交給 spring session 來管了。接着在代碼中，就用原生的 session 操作，就是直接基於 spring sesion 從 redis 中獲取數據了。
實現分布式的會話有很多種方式，我說的只不過是比較常見的幾種方式，tomcat + redis 早期比較常用，但是會重耦合到 tomcat 中；近些年，通過 spring session 來實現。
二、分布式事務
當我們的單個數據庫的性能產生瓶頸的時候，我們可能會對數據庫進行分區，這里所說的分區指的是物理分區，分區之后可能不同的庫就處於不同的服務器上了，這個時候單個數據庫的ACID已經不能適應這種情況了，而在這種ACID的集群環境下，再想保證集群的ACID幾乎是很難達到，或者即使能達到那么效率和性能會大幅下降，最為關鍵的是再很難擴展新的分區了，這個時候如果再追求集群的ACID會導致我們的系統變得很差，這時我們就需要引入一個新的理論原則來適應這種集群的情況，就是 CAP 原則或者叫CAP定理，那么CAP定理指的是什么呢？
CAP定理
CAP定理是由加州大學伯克利分校Eric Brewer教授提出來的，他指出WEB服務無法同時滿足一下3個屬性：
- 一致性(Consistency) ： 客戶端知道一系列的操作都會同時發生(生效)
- 可用性(Availability) ： 每個操作都必須以可預期的響應結束
- 分區容錯性(Partition tolerance) ： 即使出現單個組件無法可用,操作依然可以完成
具體地講在分布式系統中，在任何數據庫設計中，一個Web應用至多只能同時支持上面的兩個屬性。顯然，任何橫向擴展策略都要依賴於數據分區。因此，設計人員必須在一致性與可用性之間做出選擇。
這個定理在迄今為止的分布式系統中都是適用的！ 為什么這么說呢？
這個時候有同學可能會把數據庫的2PC（兩階段提交）搬出來說話了。OK，我們就來看一下數據庫的兩階段提交。
對數據庫分布式事務有了解的同學一定知道數據庫支持的2PC，又叫做 XA Transactions。
其中，XA 是一個兩階段提交協議，該協議分為以下兩個階段：
- 第一階段：事務協調器要求每個涉及到事務的數據庫預提交(precommit)此操作，並反映是否可以提交.
- 第二階段：事務協調器要求每個數據庫提交數據。
其中，如果有任何一個數據庫否決此次提交，那么所有數據庫都會被要求回滾它們在此事務中的那部分信息。這樣做的缺陷是什么呢? 咋看之下我們可以在數據庫分區之間獲得一致性。
如果CAP 定理是對的，那么它一定會影響到可用性。
如果說系統的可用性代表的是執行某項操作相關所有組件的可用性的和。那么在兩階段提交的過程中，可用性就代表了涉及到的每一個數據庫中可用性的和。我們假設兩階段提交的過程中每一個數據庫都具有99.9%的可用性，那么如果兩階段提交涉及到兩個數據庫，這個結果就是99.8%。根據系統可用性計算公式，假設每個月43200分鍾，99.9%的可用性就是43157分鍾, 99.8%的可用性就是43114分鍾，相當於每個月的宕機時間增加了43分鍾。
以上，可以驗證出來，CAP定理從理論上來講是正確的，CAP我們先看到這里，等會再接着說。
在分布式系統中，要實現分布式事務，無外乎那幾種解決方案。
分布式事務的實現主要有以下 5 種方案：
- XA 方案
- TCC 方案
- 本地消息表
- 可靠消息最終一致性方案
- 最大努力通知方案
兩階段提交方案/XA方案
所謂的 XA 方案，即：兩階段提交，有一個事務管理器的概念，負責協調多個數據庫（資源管理器）的事務，事務管理器先問問各個數據庫你准備好了嗎？如果每個數據庫都回復 ok，那么就正式提交事務，在各個數據庫上執行操作；如果任何其中一個數據庫回答不 ok，那么就回滾事務。
這種分布式事務方案，比較適合單塊應用里，跨多個庫的分布式事務，而且因為嚴重依賴於數據庫層面來搞定復雜的事務，效率很低，絕對不適合高並發的場景。如果要玩兒，那么基於 Spring + JTA 就可以搞定，自己隨便搜個 demo 看看就知道了。
這個方案，我們很少用，一般來說某個系統內部如果出現跨多個庫的這么一個操作，是不合規的。我可以給大家介紹一下， 現在微服務，一個大的系統分成幾十個甚至幾百個服務。一般來說，我們的規定和規范，是要求每個服務只能操作自己對應的一個數據庫。
如果你要操作別的服務對應的庫，不允許直連別的服務的庫，違反微服務架構的規范，你隨便交叉胡亂訪問，幾百個服務的話，全體亂套，這樣的一套服務是沒法管理的，沒法治理的，可能會出現數據被別人改錯，自己的庫被別人寫掛等情況。
如果你要操作別人的服務的庫，你必須是通過調用別的服務的接口來實現，絕對不允許交叉訪問別人的數據庫。
TCC 方案
TCC 的全稱是：Try、Confirm、Cancel。
- Try 階段：這個階段說的是對各個服務的資源做檢測以及對資源進行鎖定或者預留。
- Confirm 階段：這個階段說的是在各個服務中執行實際的操作。
- Cancel 階段：如果任何一個服務的業務方法執行出錯，那么這里就需要進行補償，就是執行已經執行成功的業務邏輯的回滾操作。（把那些執行成功的回滾）
這種方案說實話幾乎很少人使用，我們用的也比較少，但是也有使用的場景。因為這個事務回滾實際上是嚴重依賴於你自己寫代碼來回滾和補償了，會造成補償代碼巨大，非常之惡心。
比如說我們，一般來說跟錢相關的，跟錢打交道的，支付、交易相關的場景，我們會用 TCC，嚴格保證分布式事務要么全部成功，要么全部自動回滾，嚴格保證資金的正確性，保證在資金上不會出現問題。
而且最好是你的各個業務執行的時間都比較短。
但是說實話，一般盡量別這么搞，自己手寫回滾邏輯，或者是補償邏輯，實在太惡心了，那個業務代碼是很難維護的。
本地消息表
本地消息表其實是國外的 ebay 搞出來的這么一套思想。
這個大概意思是這樣的：
- A 系統在自己本地一個事務里操作同時，插入一條數據到消息表；
- 接着 A 系統將這個消息發送到 MQ 中去；
- B 系統接收到消息之后，在一個事務里，往自己本地消息表里插入一條數據，同時執行其他的業務操作，如果這個消息已經被處理過了，那么此時這個事務會回滾，這樣保證不會重復處理消息；
- B 系統執行成功之后，就會更新自己本地消息表的狀態以及 A 系統消息表的狀態；
- 如果 B 系統處理失敗了，那么就不會更新消息表狀態，那么此時 A 系統會定時掃描自己的消息表，如果有未處理的消息，會再次發送到 MQ 中去，讓 B 再次處理；
- 這個方案保證了最終一致性，哪怕 B 事務失敗了，但是 A 會不斷重發消息，直到 B 那邊成功為止。
這個方案說實話最大的問題就在於嚴重依賴於數據庫的消息表來管理事務啥的，如果是高並發場景咋辦呢？咋擴展呢？所以一般確實很少用。
可靠消息最終一致性方案
這個的意思，就是干脆不要用本地的消息表了，直接基於 MQ 來實現事務。比如阿里的 RocketMQ 就支持消息事務。
大概的意思就是：
- A 系統先發送一個 prepared 消息到 mq，如果這個 prepared 消息發送失敗那么就直接取消操作別執行了；
- 如果這個消息發送成功過了，那么接着執行本地事務，如果成功就告訴 mq 發送確認消息，如果失敗就告訴 mq 回滾消息；
- 如果發送了確認消息，那么此時 B 系統會接收到確認消息，然后執行本地的事務；
- mq 會自動定時輪詢所有 prepared 消息回調你的接口，問你，這個消息是不是本地事務處理失敗了，所有沒發送確認的消息，是繼續重試還是回滾？一般來說這里你就可以查下數據庫看之前本地事務是否執行，如果回滾了，那么這里也回滾吧。這個就是避免可能本地事務執行成功了，而確認消息卻發送失敗了。
- 這個方案里，要是系統 B 的事務失敗了咋辦？重試咯，自動不斷重試直到成功，如果實在是不行，要么就是針對重要的資金類業務進行回滾，比如 B 系統本地回滾后，想辦法通知系統 A 也回滾；或者是發送報警由人工來手工回滾和補償。
- 這個還是比較合適的，目前國內互聯網公司大都是這么玩兒的，要不你舉用 RocketMQ 支持的，要不你就自己基於類似 ActiveMQ？RabbitMQ？自己封裝一套類似的邏輯出來，總之思路就是這樣子的。
最大努力通知方案
這個方案的大致意思就是：
- 系統 A 本地事務執行完之后，發送個消息到 MQ；
- 這里會有個專門消費 MQ 的最大努力通知服務，這個服務會消費 MQ 然后寫入數據庫中記錄下來，或者是放入個內存隊列也可以，接着調用系統 B 的接口；
- 要是系統 B 執行成功就 ok 了；要是系統 B 執行失敗了，那么最大努力通知服務就定時嘗試重新調用系統 B，反復 N 次，最后還是不行就放棄。
你們公司是如何處理分布式事務的？
如果你真的被問到，可以這么說，我們某某特別嚴格的場景，用的是 TCC 來保證強一致性；然后其他的一些場景基於阿里的 RocketMQ 來實現分布式事務。
你找一個嚴格資金要求絕對不能錯的場景，你可以說你是用的 TCC 方案；如果是一般的分布式事務場景，訂單插入之后要調用庫存服務更新庫存，庫存數據沒有資金那么的敏感，可以用可靠消息最終一致性方案。
友情提示一下，RocketMQ 3.2.6 之前的版本，是可以按照上面的思路來的，但是之后接口做了一些改變，我這里不再贅述了。
當然如果你願意，你可以參考可靠消息最終一致性方案來自己實現一套分布式事務，比如基於 RocketMQ 來玩兒。
三、分布式鎖
redis 分布式鎖
redis 最普通的分布式鎖
第一個最普通的實現方式，就是在 redis 里創建一個 key，這樣就算加鎖。
SET my:lock 隨機值 NX PX 30000
執行這個命令就 ok。
NX：表示只有key不存在的時候才會設置成功。（如果此時 redis 中存在這個 key，那么設置失敗，返回nil）PX 30000：意思是 30s 后鎖自動釋放。別人創建的時候如果發現已經有了就不能加鎖了。
釋放鎖就是刪除 key ，但是一般可以用 lua 腳本刪除，判斷 value 一樣才刪除：
-- 刪除鎖的時候，找到 key 對應的 value，跟自己傳過去的 value 做比較，如果是一樣的才刪除。 if redis.call("get",KEYS[1]) == ARGV[1] then return redis.call("del",KEYS[1]) else return 0 end
為啥要用隨機值呢？因為如果某個客戶端獲取到了鎖，但是阻塞了很長時間才執行完，比如說超過了 30s，此時可能已經自動釋放鎖了，此時可能別的客戶端已經獲取到了這個鎖，要是你這個時候直接刪除 key 的話會有問題，所以得用隨機值加上面的 lua 腳本來釋放鎖。
但是這樣是肯定不行的。因為如果是普通的 redis 單實例，那就是單點故障。或者是 redis 普通主從，那 redis 主從異步復制，如果主節點掛了（key 就沒有了），key 還沒同步到從節點，此時從節點切換為主節點，別人就可以 set key，從而拿到鎖。
RedLock 算法
這個場景是假設有一個 redis cluster，有 5 個 redis master 實例。然后執行如下步驟獲取一把鎖：
- 獲取當前時間戳，單位是毫秒；
- 跟上面類似，輪流嘗試在每個 master 節點上創建鎖，過期時間較短，一般就幾十毫秒；
- 嘗試在大多數節點上建立一個鎖，比如 5 個節點就要求是 3 個節點
n / 2 + 1； - 客戶端計算建立好鎖的時間，如果建立鎖的時間小於超時時間，就算建立成功了；
- 要是鎖建立失敗了，那么就依次之前建立過的鎖刪除；
- 只要別人建立了一把分布式鎖，你就得不斷輪詢去嘗試獲取鎖。
zk 分布式鎖
zk 分布式鎖，其實可以做的比較簡單，就是某個節點嘗試創建臨時 znode，此時創建成功了就獲取了這個鎖；這個時候別的客戶端來創建鎖會失敗，只能注冊個監聽器監聽這個鎖。釋放鎖就是刪除這個 znode，一旦釋放掉就會通知客戶端，然后有一個等待着的客戶端就可以再次重新加鎖。
public class ZooKeeperSession { private static CountDownLatch connectedSemaphore = new CountDownLatch(1); private ZooKeeper zookeeper; private CountDownLatch latch; public ZooKeeperSession() { try { this.zookeeper = new ZooKeeper("192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181", 50000, new ZooKeeperWatcher()); try { connectedSemaphore.await(); } catch (InterruptedException e) { e.printStackTrace(); } System.out.println("ZooKeeper session established......"); } catch (Exception e) { e.printStackTrace(); } } /** * 獲取分布式鎖 * * @param productId */ public Boolean acquireDistributedLock(Long productId) { String path = "/product-lock-" + productId; try { zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); return true; } catch (Exception e) { while (true) { try { // 相當於是給node注冊一個監聽器，去看看這個監聽器是否存在 Stat stat = zk.exists(path, true); if (stat != null) { this.latch = new CountDownLatch(1); this.latch.await(waitTime, TimeUnit.MILLISECONDS); this.latch = null; } zookeeper.create(path, "".getBytes(), Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL); return true; } catch (Exception ee) { continue; } } } return true; } /** * 釋放掉一個分布式鎖 * * @param productId */ public void releaseDistributedLock(Long productId) { String path = "/product-lock-" + productId; try { zookeeper.delete(path, -1); System.out.println("release the lock for product[id=" + productId + "]......"); } catch (Exception e) { e.printStackTrace(); } } /** * 建立zk session的watcher * * @author bingo * @since 2018/11/29 * */ private class ZooKeeperWatcher implements Watcher { public void process(WatchedEvent event) { System.out.println("Receive watched event: " + event.getState()); if (KeeperState.SyncConnected == event.getState()) { connectedSemaphore.countDown(); } if (this.latch != null) { this.latch.countDown(); } } } /** * 封裝單例的靜態內部類 * * @author bingo * @since 2018/11/29 * */ private static class Singleton { private static ZooKeeperSession instance; static { instance = new ZooKeeperSession(); } public static ZooKeeperSession getInstance() { return instance; } } /** * 獲取單例 * * @return */ public static ZooKeeperSession getInstance() { return Singleton.getInstance(); } /** * 初始化單例的便捷方法 */ public static void init() { getInstance(); } }
也可以采用另一種方式，創建臨時順序節點：
如果有一把鎖，被多個人給競爭，此時多個人會排隊，第一個拿到鎖的人會執行，然后釋放鎖；后面的每個人都會去監聽排在自己前面的那個人創建的 node 上，一旦某個人釋放了鎖，排在自己后面的人就會被 zookeeper 給通知，一旦被通知了之后，就 ok 了，自己就獲取到了鎖，就可以執行代碼了。
public class ZooKeeperDistributedLock implements Watcher { private ZooKeeper zk; private String locksRoot = "/locks"; private String productId; private String waitNode; private String lockNode; private CountDownLatch latch; private CountDownLatch connectedLatch = new CountDownLatch(1); private int sessionTimeout = 30000; public ZooKeeperDistributedLock(String productId) { this.productId = productId; try { String address = "192.168.31.187:2181,192.168.31.19:2181,192.168.31.227:2181"; zk = new ZooKeeper(address, sessionTimeout, this); connectedLatch.await(); } catch (IOException e) { throw new LockException(e); } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } } public void process(WatchedEvent event) { if (event.getState() == KeeperState.SyncConnected) { connectedLatch.countDown(); return; } if (this.latch != null) { this.latch.countDown(); } } public void acquireDistributedLock() { try { if (this.tryLock()) { return; } else { waitForLock(waitNode, sessionTimeout); } } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } } public boolean tryLock() { try { // 傳入進去的locksRoot + “/” + productId // 假設productId代表了一個商品id，比如說1 // locksRoot = locks // /locks/10000000000，/locks/10000000001，/locks/10000000002 lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL); // 看看剛創建的節點是不是最小的節點 // locks：10000000000，10000000001，10000000002 List<String> locks = zk.getChildren(locksRoot, false); Collections.sort(locks); if(lockNode.equals(locksRoot+"/"+ locks.get(0))){ //如果是最小的節點,則表示取得鎖 return true; } //如果不是最小的節點，找到比自己小1的節點 int previousLockIndex = -1; for(int i = 0; i < locks.size(); i++) { if(lockNode.equals(locksRoot + “/” + locks.get(i))) { previousLockIndex = i - 1; break; } } this.waitNode = locks.get(previousLockIndex); } catch (KeeperException e) { throw new LockException(e); } catch (InterruptedException e) { throw new LockException(e); } return false; } private boolean waitForLock(String waitNode, long waitTime) throws InterruptedException, KeeperException { Stat stat = zk.exists(locksRoot + "/" + waitNode, true); if (stat != null) { this.latch = new CountDownLatch(1); this.latch.await(waitTime, TimeUnit.MILLISECONDS); this.latch = null; } return true; } public void unlock() { try { // 刪除/locks/10000000000節點 // 刪除/locks/10000000001節點 System.out.println("unlock " + lockNode); zk.delete(lockNode, -1); lockNode = null; zk.close(); } catch (InterruptedException e) { e.printStackTrace(); } catch (KeeperException e) { e.printStackTrace(); } } public class LockException extends RuntimeException { private static final long serialVersionUID = 1L; public LockException(String e) { super(e); } public LockException(Exception e) { super(e); } } }
redis 分布式鎖和 zk 分布式鎖的對比
- redis 分布式鎖，其實需要自己不斷去嘗試獲取鎖，比較消耗性能。
- zk 分布式鎖，獲取不到鎖，注冊個監聽器即可，不需要不斷主動嘗試獲取鎖，性能開銷較小。
另外一點就是，如果是 redis 獲取鎖的那個客戶端 出現 bug 掛了，那么只能等待超時時間之后才能釋放鎖；而 zk 的話，因為創建的是臨時 znode，只要客戶端掛了，znode 就沒了，此時就自動釋放鎖。
redis 分布式鎖大家沒發現好麻煩嗎？遍歷上鎖，計算時間等等......zk 的分布式鎖語義清晰實現簡單。
所以先不分析太多的東西，就說這兩點，我個人實踐認為 zk 的分布式鎖比 redis 的分布式鎖牢靠、而且模型簡單易用。
四、接口冪等性