FolkMQ v1.2 开源消息中间件（添加事务消息）

内存型消息中间件（单机版，约 180K TPS）。"单线程"、"多路复用"、"快照持久化" 和 "Broker 模式集群"！支持定时消息，过期消息，顺序消息，事务消息，请求消息。支持确认、重试、延时。支持 Qos0、Qos1。

组合缺一

688人浏览 · 2024-02-27 16:55:01

组合缺一 · 2024-02-27 16:55:01 发布

简介

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

启动服务：

docker run -p 18602:18602 -p 8602:8602 noearorg/folkmq-server:1.2.2

新功能示例（事务消息）：

//准备（1.取名字；2.添加事务回查）
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);
    }
});
    
//发送事务消息    
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();
}

新功能示例（发送与监听模式 - rpc）：

//客户端1
MqClient client1 = FolkMQ.createClient("folkmq://127.0.0.1:18602")
                        .nameAs("demoapp1")
                        .connect();

//客户端1监听
client1.listen(new MqRouter(m -> m.getTag()).doOn("hello", request -> {
    System.out.println(request);
    request.acknowledge(new StringEntity("me to!"));
}));


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

//客户端2发送
Reply reply = client2.send(new MqMessage("helloworld!").tag("hello"), "demoapp1").await();
System.out.println(reply.dataAsString());