FolkMQ 是怎样进行消息的事务处理?

来源: 投稿
2024-03-02 21:35:00

FolkMQ 提供了二段式提交的事务提交的机制(TCC 模型)。允许生产者在发送消息时绑定到一个事务中并接收事务的管理,以确保消息的原子性(要么全成功,要么全失败)。在 FolkMQ 中,事务是通过 MqTransaction 接口实现管理的。一般是通过:

  1. 新建事务:在产生者端,可以通过调用 MqTransaction tran = client.newTransaction() 新建一个事务。后续的消息发送与此事务绑定。绑定的消息,即参与此事务;没绑定的消息,则照旧。
  2. 发送消息(T):接下来的事务中,尝试发送消息。此时,中间件会把消息放在一个事务中转队列里。
  3. 提交事务(C):如果所有消息都发送成功了,就通过 tran.commit() 方法提交事务。此时,中间件把事务相关消息从事务中转队列取出,并转到派发队列。
  4. 回滚事务(C):通过 try catch,如果有某一条消息发送异常了。就通过 tran.rollback() 方法回滚事务,即所有消息全部取消。此时,中间件把事务相关消息从事务中转队列删掉。

发送事务消息:

//发送事务消息    
MqTransaction tran = client.newTransaction();

try {
    //同步
    client.publish("demo", new MqMessage("demo1").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo2").attr("orderId","1").transaction(tran));
    //异步,也行!
    client.publishAsync("demo", new MqMessage("demo3").attr("orderId","1").transaction(tran));
    client.publishAsync("demo", new MqMessage("demo4").attr("orderId","1").transaction(tran));

    tran.commit();
} catch (Throwable e) {
    tran.rollback();
}

FolkMQ 的服务端事务处理,是基于专属的"事务中转队列"实现(支持快照持久化)。它可以保证在事务中的消息,要么全部成功,要么全部失败。而且,当客户端没有提交二次确认时(偶尔会有网络原因),超过60秒后,会发起反向“回查”。要实现更“周密”的事务处理,客户端还需要实现回查处理。一般在客户端初始化时,或者连接建立后配置事务回查处理:

MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名
    .connect();

client.transactionCheckback(m->{
  if("1".equals(m.getAttr("orderId"))) {
      m.acknowledge(true);
  }
});

FolkMQ 的事务机制,并且不会对性能有影响,与非事务消息性能没什么差别。

展开阅读全文
点击引领话题📣 发布并加入讨论🔥
0 评论
1 收藏
分享
返回顶部
顶部