分布式程序员必读：你与分布式和事务真的很熟吗？

微吐槽

hello,world.

不想了，我等码农，还是看看怎么来处理分布式系统中的事务这个老大难吧!

• 本文略长，读者需要有一定耐心，如果你是高级码农或者架构师级别，你可以跳过。
• 本文注重实战或者实现，不涉及CAP，略提ACID。
• 本文适合基础分布式程序员：

1. 本文会涉及集群中节点的failover和recover问题.
2. 本文会涉及事务及不透明事务的问题.
3. 本文会提到微博和tweeter，并引出一个大数据问题.

由于分布式这个话题太大，事务这个话题也太大，我们从一个集群的一个小小节点开始谈起。

集群中存活的节点与同步

分布式系统中，如何判断一个节点(node)是否存活?

kafka这样认为：

1. 此节点和zookeeper能喊话.(Keep sessions with zookeeper through heartbeats.)
2. 此节点如果是个从节点，必须能够尽可能忠实地反映主节点的数据变化。

也就是说，必须能够在主节点写了新数据后，及时复制这些变化的数据，所谓及时，不能拉下太多哦.

那么，符合上面两个条件的节点就可以认为是存活的，也可以认为是同步的(in-sync).

关于第1点，大家对心跳都很熟悉，那么我们可以这样认为某个节点不能和zookeeper喊话了：

1. zookeeper-node: 
2. var timer =  
3. new timer() 
4. .setInterval(10sec) 
5. .onTime(slave-nodes,function(slave-nodes){ 
6.     slave-nodes.forEach( node -> { 
7.         boolean isAlive = node.heartbeatACK(15sec); 
8.         if(!isAlive) { 
9.             node.numNotAlive += 1; 
10.             if(node.numNotAlive >= 3) { 
11.                 node.declareDeadOrFailed(); 
12.                 slave-nodes.remove(node); 
13.                  
14.                 //回调也可 leader-node-app.notifyNodeDeadOrFailed(node) 
15.                  
16.             } 
17.         }else  
18.         node.numNotAlive = 0; 
19.     }); 
20. }); 
21.  
22. timer.run(); 
23.  
24. //你可以回调也可以像下面这样简单的计时判断 
25. leader-node-app: 
26. var timer =  
27. new timer() 
28. .setInterval(10sec) 
29. .onTime(slave-nodes,function(slave-nodes){ 
30.     slave-nodes.forEach(node -> { 
31.         if(node.isDeadOrFailed) { 
32.  
33.         //node不能和zookeeper喊话了 
34.              
35.         } 
36.     }); 
37. }); 
38.  
39. timer.run(); 

关于第二点，要稍微复杂点了，怎么搞呢?

来这么分析：

• 数据 messages.
• 操作 op-log.
• 偏移 position/offset.

1. // 1. 先考虑messages 
2. // 2. 再考虑log的postion或者offset 
3. // 3. 考虑msg和off都记录在同源数据库或者存储设备上.(database or storage-device.) 
4. var timer =  
5. new timer() 
6. .setInterval(10sec) 
7. .onTime(slave-nodes,function(nodes){ 
8.     var core-of-cpu = 8; 
9.     //嫌慢就并发呗 mod hash go! 
10.     nodes.groupParallel(core-of-cpu) 
11.     .forEach(node -> { 
12.         boolean nodeSucked = false; 
13.  
14.         if(node.ackTimeDiff > 30sec) { 
15.             //30秒内没有回复，node卡住了 
16.             nodeSucked = true; 
17.         } 
18.         if(node.logOffsetDiff > 100) { 
19.             //node复制跟不上了，差距超过100条数据 
20.             nodeSucked = true; 
21.         } 
22.  
23.         if(nodeSucked) { 
24.             //总之node“死”掉了，其实到底死没死，谁知道呢？network-error在分布式系统中或者节点失败这个事情是正常现象. 
25.             node.declareDeadOrFailed(); 
26.             //不和你玩啦，集群不要你了 
27.             nodes.remove(node); 
28.             //该怎么处理呢，抛个事件吧. 
29.             fire-event-NodeDeadOrFailed(node); 
30.         } 
31.     }); 
32. }); 
33.  
34. timer.run(); 

上面的节点的状态管理一般由zookeeper来做，leader或者master节点也会维护那么点状态。

那么应用中的leader或者master节点，只需要从zookeeper拉状态就可以，同时，上面的实现是不是一定最佳呢?不是的，而且多数操作可以合起来，但为了描述节点是否存活这个事儿，咱们这么写没啥问题。

节点死掉、失败、不同步了，咋处理呢?

好嘛，终于说到failover和recover了，那failover比较简单，因为还有其它的slave节点在，不影响数据读取。

1. 同时多个slave节点失败了?没有100%的可用性.数据中心和机房瘫痪、网络电缆切断、hacker入侵删了你的根，总之你rp爆表了.
2. 如果主节点失败了，那master-master不行嘛?keep-alived或者LVS或者你自己写failover吧.高可用架构(HA)又是个大件儿了，此文不展开了。

我们来关注下recover方面的东西，这里把视野打开点，不仅关注slave节点重启后追log来同步数据，我们看下在实际应用中，数据请求(包括读、写、更新)失败怎么办?

大家可能都会说，重试(retry)呗、重放(replay)呗或者干脆不管了呗!

行，都行，这些都是策略，但具体怎么个搞法，你真的清楚了?

一个bigdata问题

我们先摆个探讨的背景：

怎么解决呢?

为了简单，我们忽略掉日期，先看看这个方法行不行：

1. /** --------------------------------- 
2. * 1. 求出转发微博(url)的大V.  
3. * __________________________________*/ 
4.  
5. 方法 ：getUrlToTweetersMap(String url_id) 
6.  
7. SQL ： /* 数据库A，表url_user存储了转发某url的user */ 
8. SELECT url_user.user_id as tweeter_id 
9. FROM url_user 
10. WHERE url_user.url_id = ${url_id} 
11.  
12. 返回 ：[user_1,...,user_m] 

1. /** --------------------------------- 
2. * 2. 求出大V的粉丝  
3. * __________________________________*/ 
4.  
5. 方法 : getFollowers(String tweeter_id); 
6.  
7. SQL :   /* 数据库B */ 
8. SELECT users.id as user_id 
9. FROM users 
10. WHERE users.followee_id = ${tweeter_id} 
11.  
12. 返回：tweeter的粉丝 

1. /** --------------------------------- 
2. * 3. 求出Reach 
3. * __________________________________*/ 
4.  
5. var url = queryArgs.getUrl(); 
6. var tweeters = getUrlToTweetersMap(); 
7. var result = new HashMap<String,Integer>(); 
8. tweeters.forEach(t -> { 
9.     // 你可以批量in + 并发读来优化下面方法的性能 
10.     var followers = getFollowers(t.tweeter_id); 
11.  
12.     followers.forEach(f -> { 
13.         //hash去重 
14.         result.put(f.user_id,1); 
15.     }); 
16. }); 
17.  
18. //Reach 
19. return result.size(); 

顶呱呱，无论如何，求出了Reach啊!

其实这又引出了一个很重要的问题，也是很多大谈框架、设计、模式却往往忽视的问题：性能和数据库建模的关系。

1.数据量有多大?

不知道读者有木有对这个问题的数据库I/O有点想法，或者虎躯一震呢?

Computing reach is too intense for a single machine – it can require thousands of database calls and tens of millions of tuples.

在上面的数据库设计中避免了JOIN，为了提高求大V粉丝的性能，可以将一批大V作为batch/bulk，然后多个batch并发读，誓死搞死数据库。

这里将微博到转发者表所在的库，与粉丝库分离，如果数据更大怎么办?

库再分表...

OK，假设你已经非常熟悉传统关系型数据库的分库分表及数据路由(读路径的聚合、写路径的分发)、或者你对于sharding技术也很熟悉、或者你良好的结合了HBase的横向扩展能力并有一致性策略来解决其二级索引问题.

总之，存储和读取的问题假设你已经解决了，那么分布式计算呢?

2.微博这种应用，人与人之间的关系成图状(网)，你怎么建模存储?而不仅仅对应这个问题，比如：

某人的好友的好友可能和某人有几分相熟?

看看用storm怎么来解决分布式计算，并提供流式计算的能力：

1. // url到大V -> 数据库1 
2. TridentState urlToTweeters = 
3.     topology.newStaticState(getUrlToTweetersState()); 
4. // 大V到粉丝 -> 数据库2 
5. TridentState tweetersToFollowers = 
6.     topology.newStaticState(getTweeterToFollowersState()); 
7.  
8. topology.newDRPCStream("reach") 
9.     .stateQuery(urlToTweeters, new Fields("args"), new MapGet(), new Fields("tweeters")) 
10.     .each(new Fields("tweeters"), new ExpandList(), new Fields("tweeter")) 
11.     .shuffle() /* 大V的粉丝很多，所以需要分布式处理*/ 
12.     .stateQuery(tweetersToFollowers, new Fields("tweeter"), new MapGet(), new Fields("followers")) 
13.     .parallelismHint(200) /* 粉丝很多，所以需要高并发 */  
14.     .each(new Fields("followers"), new ExpandList(), new Fields("follower")) 
15.     .groupBy(new Fields("follower")) 
16.     .aggregate(new One(), new Fields("one")) /* 去重 */ 
17.     .parallelismHint(20) 
18.     .aggregate(new Count(), new Fields("reach")); /* 计算reach数 */ 

最多处理一次(At most once)

回到主题，引出上面的例子，一是为了引出一个有关分布式(存储+计算)的问题，二是透漏这么点意思：

码农，就应该关注设计和实现的东西，比如Jay Kreps是如何发明Kafka这个轮子的 : ]

如果你还是码农级别，咱来务点实吧，前面我们说到recover，节点恢复的问题，那么我们恢复几个东西?

基本的：

• 节点状态
• 节点数据

本篇从数据上来讨论下这个问题，为使问题再简单点，我们考虑写数据的场景，如果我们用write-ahead-log的方式来保证数据复制和一致性，那么我们会怎么处理一致性问题呢?

主节点有新数据写入.

从节点追log,准备复制这批新数据。从节点做两件事：

(1). 把数据的id偏移写入log;

(2). 正要处理数据本身，从节点挂了。

那么根据上文的节点存活条件，这个从节点挂了这件事被探测到了，从节点由维护人员手动或者其自己恢复了，那么在加入集群和小伙伴们继续玩耍之前，它要同步自己的状态和数据。

问题来了：

如果根据log内的数据偏移来同步数据，那么，因为这个节点在处理数据之前就把偏移写好了，可是那批数据lost-datas没有得到处理，如果追log之后的数据来同步，那么那批数据lost-datas就丢了。

在这种情况下，就叫作数据最多处理一次，也就是说数据会丢失。

最少处理一次(At least once)

好吧，丢失数据不能容忍，那么我们换种方式来处理：

1.主节点有新数据写入.

2.从节点追log,准备复制这批新数据。从节点做两件事：

(1). 先处理数据;

(2). 正要把数据的id偏移写入log，从节点挂了。

问题又来了：

仅处理一次(Exactly once)

Transaction

好吧，数据重复也不能容忍?要求挺高啊。

大家都追求的强一致性保证(这里是最终一致性)，怎么来搞呢?

换句话说，在更新数据的时候，事务能力如何保障呢?

假设一批数据如下：

// 新到数据 { transactionId:4 urlId:99 reach:5 }

现在要更新这批数据到库里或者log里，那么原来的情况是：

// 老数据
{
transactionId：3
urlId:99
reach:3
}

如果说可以保证如下三点：

事务ID的生成是强有序的.(隔离性，串行)

同一个事务ID对应的一批数据相同.(幂等性，多次操作一个结果)

单条数据会且仅会出现在某批数据中.(一致性，无遗漏无重复)

那么，放心大胆的更新好了：

// 更新后数据
{
transactionId：4
urlId:99
//3 + 5 = 8
reach:8
}

这里是更新成功了。如果更新的时候，节点挂了，那么库里或者log里的id偏移不写，数据也不处理，等节点恢复，就可以放心去同步，然后加入集群玩耍了。

所以说，要保证数据仅处理一次，还是挺困难的吧?

上面的保障“仅处理一次”这个语义的实现有什么问题呢?

性能问题。

如果不考虑性能，就此作罢，这也不是什么大事。

你似乎意犹未尽?来吧，看看“银弹”是什么?

Opaque-Transaction

现在，我们来追求这样一种效果：

来看看例子吧，老数据不变，只是多了个字段：prevReach。

// 老数据
{
transactionId：3
urlId:99
//注意这里多了个字段，表示之前的reach的值
prevReach:2
reach:3
}


// 新到数据
{
transactionId:4
urlId:99
reach:5
}

这种情况，新事务的ID更大、更靠后，表明新事务可以执行，还等什么，直接更新，更新后数据如下：

// 新到数据
{
transactionId:4
urlId:99
//注意这里更新为之前的值
prevReach:3
//3 + 5 = 8
reach:8
}

现在来看下另外的情况：

// 老数据
{
transactionId：3
urlId:99
prevReach:2
reach:3
}

// 新到数据
{
//注意事务ID为3，和老数据中的事务ID相同
transactionId:3
urlId:99
reach:5
}

这种情况怎么处理?是跳过吗?因为新数据的事务ID和库里或者log里的事务ID相同，按事务要求这次数据应该已经处理过了，跳过?

不，这种事不能靠猜的，想想我们有的几个性质，其中关键一点就是：

我们应该这么做，考虑到新到数据的事务ID和存储中的事务ID一致，所以这批数据可能被分别或者异步处理了，但是，这批数据对应的事务ID永远是同一个，那么，即使这批数据中的A部分先处理了，由于大家都是一个事务ID，那么A部分的前值是可靠的。

所以，我们将依靠prevReach而不是Reach的值来更新：

// 更新后数据
{
transactionId:3
urlId:99
//这个值不变
prevReach:2
//2 + 5 = 7
reach:7
}

你发现了什么呢?

不同的事务ID，导致了不同的值：

当事务ID为4，大于存储中的事务ID3，Reach更新为3+5 = 8.

当事务ID为3，等于存储中的事务ID3，Reach更新为2+5 = 7.

这就是Opaque Transaction.

这种事务能力是最强的了，可以保证事务异步提交。所以不用担心被卡住了，如果说集群中：

Transaction：

• 数据是分批处理的，每个事务ID对应一批确定、相同的数据.
• 保证事务ID的产生是强有序的.
• 保证分批的数据不重复、不遗漏.
• 如果事务失败，数据源丢失，那么后续事务就卡住直到数据源恢复.

Opaque-Transaction：

1. 数据是分批处理的，每批数据有确定而唯一的事务ID.
2. 保证事务ID的产生是强有序的.
3. 保证分批的数据不重复、不遗漏.
4. 如果事务失败，数据源丢失，不影响后续事务，除非后续事务的数据源也丢了.

其实这个全局ID的设计也是门艺术：

• 冗余关联表的ID，以减少join，做到O(1)取ID.
• 冗余日期(long型)字段，以避免order by.
• 冗余过滤字段，以避免无二级索引(HBase)的尴尬.
• 存储mod-hash的值，以方便分库、分表后，应用层的数据路由书写.

这个内容也太多，话题也太大，就不在此展开了。

你现在知道twitter的snowflake生成全局唯一且有序的ID的重要性了。

两阶段提交

现在用zookeeper来做两阶段提交已经是入门级技术，所以也不展开了。

如果你的数据库不支持原子操作，那么考虑两阶段提交吧。

本文作者：foreach_break

来源：51CTO