Transactional spouts
``
if(database txid=current txid){//两次更新的txid相同
跳过;
}else{
用current value替换掉database value;
}
Batch State Update
1 {SF:320:378911 = 4}
2 {SF:320:378911 = 10}
3 {SF:320:378911 = 8}
Batch State Update
1 { Batch = 1 } {SF:320:378911 = 4}
2 { Batch = 2 } {SF:320:378911 = 14}
3 { Batch = 3 } {SF:320:378911 = 22}
3 (Replayed) { Batch = 3 } {SF:320:378911 = 22}
``
Opaque Transactional spouts
``
if(database txid=current txid){
value=preValue+current value；//重新更新value
//preValue不变；
}else{
preValue=value;//更新preValue
value=preValue+current value;//更新value
txid=current txid;//更新txid
}
Batch State Update
1 {SF:320:378911 = 4}
2 {SF:320:378911 = 10}
3 {SF:320:378911 = 8}
3(Replayed) {SF:320:378911 = 6}
Completed batch Batch committed Previous state Current state
1 1 {} { SF:320:378911 = 4 }
2 2 { SF:320:378911 = 4 } { SF:320:378911 = 14 }
3 (Applies) 3 { SF:320:378911 = 14 } { SF:320:378911 = 22 }
3 (Replayed)) 3 { SF:320:378911 = 14 } { SF:320:378911 = 20 }
``
