我要评论apache omid tso 组件实现原理
作用
独立进程，处理全局事务之间的并发冲突。
流程
tsochannelhandler#channelread -> abstractrequestprocessor -> persistenceprocessorhandler
总体流程
thread1
tsochannelhandler#channelread
abstractrequestprocessor#timestamprequest 接收 client 请求，创建 requestevent 并 publish
thread2
abstractrequestprocessor#onevent 处理 requestevent 请求
abstractrequestprocessor#handlerequest
persistenceprocessorimpl#addtimestamptobatch 创建 persistevent,当 batch 满了发送事件
thread3
persistenceprocessorhandler#onevent 持久化事件处理类
tsochannelhandler
继承自 netty 的 channelinboundhandleradapter，用于处理 tso 的入站请求。
channelread
委托 requestprocessor 创建 timestamprequest 和 commitrequest 请求事件。
abstractrequestprocessor
处理 timestamp 和 commit 事件。
onevent
处理 requestevent 事件，按照事件类型派发给 handletimestamp 和 handlecommit 方法进行处理。
handletimestamp
1.通过 timestamporacle 获取下一个时间戳；
2.persistenceprocessorimpl#addbatch 事件添加到 batch,但是后续对 timestamp 请求不会额外处理。
handlecommit
主要通过 hasconflictswithcommittedtransactions 判断 writeset 和 commithashmap 里是否有事务写冲突，如果没有则可以提交事务，分配 committimestamp。
private void handlecommit(requestevent event) throws exception {
long starttimestamp = event.getstarttimestamp(); // starttimestamp
iterable<long> writeset = event.writeset(); // 写入集，存储的是 cellids
collection<long> tableidset = event.gettableidset();
boolean iscommitretry = event.iscommitretry();
boolean nonemptywriteset = writeset.iterator().hasnext(); // 检查写集合是否为空，即事务是否有写操作
if (starttimestamp > lowwatermark &&
!hasconflictswithfences(starttimestamp, tableidset) &&
!hasconflictswithcommittedtransactions(starttimestamp, writeset)) { // 检查事务是否满足提交条件,通过 hasconflictswithcommittedtransactions 判断是否有事务写冲突
// 可以进行事务提交
long committimestamp = timestamporacle.next(); // 获取提交时间戳
optional<long> forwardnewwatermark = optional.absent();
if (nonemptywriteset) { // 写集合非空
long newlowwatermark = lowwatermark;
for (long r : writeset) { // 遍历写集合中的每个元素，更新其最新的写入时间戳，并计算新的低水位线
long removed = hashmap.putlatestwriteforcell(r, committimestamp); // 更新 cellid 对应的 committimestamp, 返回之前的 oldest committimestamp
newlowwatermark = math.max(removed, newlowwatermark); // 更新低水位线
}
if (newlowwatermark != lowwatermark) { // 更新低水位线
lowwatermark = newlowwatermark;
forwardnewwatermark = optional.of(lowwatermark);
}
}
forwardcommit(starttimestamp, committimestamp, c, event.getmonctx(), forwardnewwatermark); // 持久化 commit 请求
} else { // 事务不满足提交条件
if (iscommitretry) { // re-check if it was already committed but the client retried due to a lag replying
forwardcommitretry(starttimestamp, c, event.getmonctx()); // 若是提交重试，再次检查是否已提交以避免因响应延迟导致的重复提交
} else {
forwardabort(starttimestamp, c, event.getmonctx()); // 否则，中止事务
}
}
}commithashmap
通过 longcache 缓存 cellid -> lastcommittedtimestamp 的映射。
getlatestwriteforcell 方法：
根据 cellid 获取 lastcommittedtimestamp。
putlatestwriteforcell 方法：
更新 cellid 对应的 lastcommittedtimestamp。
longcache
缓存 cellid -> lastcommittedtimestamp 的映射。
get 和 set 操作都是先将原始 cellid 进行 hash 操作找到位置，所以可能存在冲突。
set
更新 cellid 对应的 lastcommittedtimestamp。
public long set(long key, long value) {
final int index = index(key); // cellid 取模返回下标，可能会冲突
int oldestindex = 0;
long oldestvalue = long.max_value;
for (int i = 0; i < associativity; ++i) {
int currindex = 2 * (index + i); // 计算 key 下标
if (cache[currindex] == key) { // 相同事务 cellid, 替换场景
oldestvalue = 0;
oldestindex = currindex;
break;
}
if (cache[currindex + 1] <= oldestvalue) { // 没找到相同的key.通过和 oldestvalue 比较会将最小的 timestamp 剔除
oldestvalue = cache[currindex + 1];
oldestindex = currindex;
}
}
// 替换最旧的键值对，将其更新为新的键值对
cache[oldestindex] = key;
cache[oldestindex + 1] = value;
return oldestvalue;
}get
获取 cellid 对应的 lastcommittedtimestamp,找不到则返回 0.
public long get(long key) {
final int index = index(key);
for (int i = 0; i < associativity; ++i) { // associativity 里存储的元素key应该是相同的
int currindex = 2 * (index + i); // 计算 key 的下标
if (cache[currindex] == key) { // 找到 cache key
return cache[currindex + 1]; // 返回对应的 value
}
}
return 0;
}persistenceprocessorimpl
将 starttimestamp 和 committimestamp 放入 batch.
addcommittobatch
创建 event,添加到 current batch
如果 current batch is full
triggercurrentbatchflushtriggercurrentbatchflush
创建 persistbatchevent 并发送事件
persistenceprocessorhandler
处理上面 persistenceprocessorimpl 发送过来的事件，进行持久化处理。
onevent
实际上只处理 commit 事件，会创建 put 对象将事务信息持久化到 hbase 的 committable (omid_commit_table).
hbasecommittable
构造方法: 根据 hbasecommittableconfig 配置初始化
到此这篇关于apache omid tso 组件源码实现原理的文章就介绍到这了,更多相关apache omid tso 组件内容请搜索代码网以前的文章或继续浏览下面的相关文章希望大家以后多多支持代码网！
发表评论