我们都是架构师！
关注架构师(JiaGouX)，添加“星标”
获取每天技术干货，一起成为牛逼架构师
技术群请加若飞：1321113940 进架构师群
投稿、合作、版权等邮箱：[email protected]
分布式之分布式事务、分布式锁、分布式session
原文：www.cnblogs.com/heqiyoujing/p/10917102.html
一、分布式session
jsessionid cookie，就根据这个东西，在服务端可以维护一个对应的 session 域，里面可以放点数据。
Tomcat RedisSessionManager 的东西，让所有我们部署的 tomcat 都将 session 数据存储到 redis 即可。
<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"/>
<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>
示例代码：
public class TestController {public String putIntoSession(HttpServletRequest request, String username) {request.getSession().setAttribute("name", "leo");return "ok";}public String getFromSession(HttpServletRequest request, Model model){String name = request.getSession().getAttribute("name");return name;}}
二、分布式事务
CAP定理
-
一致性(Consistency) ：客户端知道一系列的操作都会同时发生(生效) -
可用性(Availability) ：每个操作都必须以可预期的响应结束 -
分区容错性(Partition tolerance) ：即使出现单个组件无法可用,操作依然可以完成
-
第一阶段：事务协调器要求每个涉及到事务的数据库预提交(precommit)此操作，并反映是否可以提交. -
第二阶段：事务协调器要求每个数据库提交数据。
-
XA 方案 -
TCC 方案 -
本地消息表 -
可靠消息最终一致性方案 -
最大努力通知方案
两阶段提交方案/XA方案
Spring + JTA 就可以搞定，自己随便搜个 demo 看看就知道了。
TCC 方案
Try、
Confirm、
Cancel。
-
Try 阶段：这个阶段说的是对各个服务的资源做检测以及对资源进行锁定或者预留。 -
Confirm 阶段：这个阶段说的是在各个服务中执行实际的操作。 -
Cancel 阶段：如果任何一个服务的业务方法执行出错，那么这里就需要进行补偿，就是执行已经执行成功的业务逻辑的回滚操作。（把那些执行成功的回滚）
本地消息表
本地消息表其实是国外的 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] thenreturn redis.call("del",KEYS[1])elsereturn 0end
为啥要用随机值呢？因为如果某个客户端获取到了锁，但是阻塞了很长时间才执行完，比如说超过了 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/10000000002lockNode = zk.create(locksRoot + "/" + productId, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);// 看看刚创建的节点是不是最小的节点// locks：10000000000，10000000001，10000000002List<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 的分布式锁牢靠、而且模型简单易用。
程序员的眼里，不止有代码和bug，还有诗与远方和妹子！！！
·END·