🎉FolkMQ v1.2 发布（添加事务消息）

2023 大语言模型技术报告.pdf

简介

采用 “单线程” + “多路复用” + "内存运行" + "快照持久化" + "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());