🎉FolkMQ v1.2 发布(添加事务消息)

来源: 投稿
2024-02-23 17:38:00

简介

  • 采用 “单线程” + “多路复用” + "内存运行" + "快照持久化" + "Broker 集群模式"(可选)
  • 基于 Socket.D 网络应用协议 开发。全新设计,自主架构!
角色 功能
生产端 发布消息、定时消息、顺序消息、可过期消息、事务消息。发送请求(rpc)。支持 Qos0、Qos1
   
消费端 订阅、取消订阅。响应请求
消费者 消费-ACK(自动、手动)
   
服务端 发布-Confirm、订阅-Confirm、取消订阅-Confirm、派发-Retry、派发-Delayed
服务端 单线程、内存运行、快照持久化(自动、停机、手动)、Broker 模式集群、集群热扩展

特点

  • 高吞吐量、低延迟

FolkMQ 纯内存运行,每秒能处理几十万条消息,最低延迟不到1毫秒。

  • 可扩展性

FolkMQ Broker 集群支持 folkmq-server 节点热扩展。流量高时随时加,流量低时可减

  • 持久性、可靠性

消息被快照持久化(类似于 redis)到本地磁盘,并且支持数据备份防止数据丢失

  • 高可用

集群内任何节点坏掉只要还有“一个”同类节点,仍可提供服务

  • 快(单机版,180K TPS)。

有点像 Redis 之于 MySql。

本次更新

  • 添加 协议版本的握手传递
  • 添加 消息事务支持(即二段式提交),支持反向事务确认
  • 添加 请求响应模式支持(即 rpc 模式)
  • 添加 消息用户属性支持
  • 优化 内存占用与快照大小
  • 优化 安全停止延时改为4秒
  • 优化 客户端相关参数校验
  • 优化 客户端的心跳间隔为6秒
  • 优化 停止打印信息
  • sokcet.d 升为 2.4.3

新功能示例(事务消息):

//准备(1.取名字;2.添加响应实现)
MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名(也当默认消息者组用)
    .connect();

//用于响应服务端发起的反向确认
client.response(m->{
  if (m.isTransaction()) {
      //极端特殊的情况下,客户端未完成事务确认。由服务端发起补尝确认
      if("1".equals(m.getAttr("orderId"))) {
          //一般这里,需要查询数据库之类的
          m.acknowledge(true);
      }
  }
});
    
//发送事务消息    
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.publish("demo", new MqMessage("demo3").attr("orderId","1").transaction(tran));
    client.publish("demo", new MqMessage("demo4").attr("orderId","1").transaction(tran));

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

新功能示例(请求响应模式):

//客户端2
MqClient client1 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demo-app1") 
    .connect();

//要支持 rpc 响应,要添加响应实现(MqResponseRouter 带了主体路由功能)
client1.response(new MqResponseRouter().doOn("test.hello", m -> {
    m.acknowledge(true, new StringEntity(m.getSender() +  ": me to! rev: " + m.getContent()));
}));

//客户端2
MqClient client2 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demo-app2")
    .connect();

//发起请求并等响应,同步模式
Reply reply = client2.request("demo-app1", "test.hello", new MqMessage("hello")).await();
print(reply.dataAsString());

代码仓库

官网

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