你真的很熟分布式处理和事务吗？，事务
微吐槽
简历上项目经历写一句hello world真的不行嘛？
不想了，我等码农，还是看看怎么来处理分布式系统中的事务这个老大难吧！
本文略长，读者需要有一定耐心，如果你是高级码农或者架构师级别，你可以跳过。
本文注重实战或者实现，不涉及CAP，略提ACID。
本文适合基础分布式程序员：
由于分布式这个话题太大，事务这个话题也太大，我们从一个集群的一个小小节点开始谈起。
集群中存活的节点与同步
分布式系统中，如何判断一个节点（node）是否存活？
kafka这样认为：
那么，符合上面两个条件的节点就可以认为是存活的，也可以认为是同步的（in-sync）.
关于第1点，大家对心跳都很熟悉，那么我们可以这样认为某个节点不能和zookeeper喊话了：
zookeeper-node:
var timer =
new timer()
.setInterval(10sec)
.onTime(slave-nodes,function(slave-nodes){
slave-nodes.forEach( node -> {
boolean isAlive = node.heatbeatACK(15sec);
if(!isAlive) {
node.numNotAlive += 1;
if(node.numNotAlive >= 3) {
node.declareDeadOrFailed();
slave-nodes.remove(node);
//回调也可 leader-node-app.notifyNodeDeadOrFailed(node)
}
}else
node.numNotAlive = 0;
});
});
timer.run();
//你可以回调也可以像下面这样简单的计时判断
leader-node-app:
var timer =
new timer()
.setInterval(10sec)
.onTime(slave-nodes,function(slave-nodes){
slave-nodes.forEach(node -> {
if(node.isDeadOrFailed) {
//node不能和zookeeper喊话了
}
});
});
timer.run();
关于第二点，要稍微复杂点了，这里也比较考究一个所谓大数据工程师或者架构师的功力了，怎么搞呢？
来这么分析：
- 数据 messages.
- 操作 op-log.
- 偏移 position/offset.
// 1. 先考虑messages
// 2. 再考虑log的postion或者offset
// 3. 考虑msg和off都记录在同源数据库或者存储设备上.(database or storage-device.)
var timer =
new timer()
.setInterval(10sec)
.onTime(slave-nodes,function(nodes){
var core-of-cpu = 8;
//嫌慢就并发呗 mod hash go!
nodes.groupParallel(core-of-cpu)
.forEach(node -> {
boolean nodeSucked = false;
if(node.ackTimeDiff > 30sec) {
//30秒内没有回复，node卡住了
nodeSucked = true;
}
if(node.logOffsetDiff > 100) {
//node复制跟不上了，差距超过100条数据
nodeSucked = true;
}
if(nodeSucked) {
//总之node“死”掉了，其实到底死没死，谁知道呢？network-error在分布式系统中或者节点失败这个事情是正常现象.
node.declareDeadOrFailed();
//不和你玩啦，集群不要你了
nodes.remove(node);
//该怎么处理呢，抛个事件吧.
fire-event-NodeDeadOrFailed(node);
}
});
});
timer.run();
上面的节点的状态管理一般由zookeeper来做，leader或者master节点也会维护那么点状态。
那么应用中的leader或者master节点，只需要从zookeeper拉状态就可以，同时，上面的实现是不是一定最佳呢？不是的，而且多数操作可以合起来，但为了描述节点是否存活这个事儿，咱们这么写没啥问题。
节点死掉、失败、不同步了，咋处理呢？
好嘛，终于说到failover和recover了，那failover比较简单，因为还有其它的slave节点在，不影响数据读取。
我们来关注下recover方面的东西，这里把视野打开点，不仅关注slave节点重启后追log来同步数据，我们看下在实际应用中，数据请求（包括读、写、更新）失败怎么办？
大家可能都会说，重试（retry）呗、重放（replay）呗或者干脆不管了呗！
行，都行，这些都是策略，但具体怎么个搞法，你真的清楚了？
一个bigdata问题
我们先摆个探讨的背景：
问题：消息流，比如微博的微博（真绕），源源不断地流进我们的应用中，要处理这些消息，有个需求是这样的：
Reach is the number of unique people exposed to a URL on Twitter.
那么，统计一下3小时内的本条微博（url）的reach总数。
怎么解决呢？
把某时间段内转发过某条微博（url）的人拉出来，把这些人的粉丝拉出来，去掉重复的人，然后求总数，就是要求的reach.
为了简单，我们忽略掉日期，先看看这个方法行不行：
/** ---------------------------------
* 1. 求出转发微博(url)的大V.
* __________________________________*/
方法 ：getUrlToTweetersMap(String url_id)
SQL ： /* 数据库A，表url_user存储了转发某url的user */
SELECT url.user_id as tweeter_id
FROM url_user
WHERE url_user.url_id = ${url_id}
返回 ：[user_1,...,user_m]
/** ---------------------------------
* 2. 求出大V的粉丝
* __________________________________*/
方法 : getFollowers(String tweeter_id);
SQL : /* 数据库B */
SELECT user.id as user_id
FROM users
WHERE users.followee_id = ${tweeter_id}
返回：tweeter的粉丝
/** ---------------------------------
* 3. 求出Reach
* __________________________________*/
var tweeters = getUrlToTweetersMap();
var result = new HashMap<String,Integer>();
tweeters.forEach(t -> {
// 你可以批量in + 并发读来优化下面方法的性能
var followers = getFollowers(t.tweeter_id);
followers.forEach(f -> {
//hash去重
result.put(f.user_id,1);
});
});
//Reach
return result.size();
顶呱呱，无论如何，求出了Reach啊！
其实这又引出了一个很重要的问题，也是很多大谈框架、设计、模式却往往忽视的问题：性能和数据库建模的关系。
其实很佩服一篇文章，几个图表，看起来很cool的架构图就把这些问题说得一清二楚，架构嘛。
看看用storm怎么来解决分布式计算，并赋予其流式计算的特性：
// url到大V -> 数据库1
TridentState urlToTweeters =
topology.newStaticState(getUrlToTweetersState());
// 大V到粉丝 -> 数据库2
TridentState tweetersToFollowers =
topology.new