Skip to content

Releases: noear/folkmq

FolkMQ v1.2.4

01 Mar 06:59
Compare
Choose a tag to compare

一、更新说明

  • 添加 folkmq-server 对 ws 输传协议的支持(方便 js 客户端使用)

新功能演示:

docker run -p 18602:18602 -p 8602:8602 -e folkmq.schema='sd:ws' noearorg/folkmq-server:1.2.4
const {FolkMQ} = require("@noear/folkmq");

async function main() {
    //创建客户端,并连接
    const client = await FolkMQ.createClient("folkmq:ws://127.0.0.1:18602")
                            .nameAs("demoapp")
                            .connect();

    //订阅主题,并指定加入的消费者分组
    client.subscribe("demo", null, true, message => {
        console.log(message);
    });

    //发布消息
    client.publish("demo", FolkMQ.newMqMessage("helloworld!"));
}

main();

二、兼容说明

  • 本次更新,向下兼容。
  • 新增的功能功能(事务消息,发送与监听模式模式,消息属性),需要新版服务端和客户端支持。

FolkMQ v1.2.3

29 Feb 09:52
Compare
Choose a tag to compare

一、更新说明

(1) for Java

  • 添加 rpc 异常传导机制

(2) for JavaScript

  • 完成 js 客户端开发(兼容:浏览器,uni-app,node.js)

(3) 新功能演示

  • 事务消息 for js:
//准备(1.取名字;2.添加事务回查)
const client = await FolkMQ.createClient("folkmq:ws://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名
    .connect();

//用于服务端发起的事务回查
client.transactionCheckback(m => {
  //极端特殊的情况下,客户端未完成事务确认。由服务端发起补尝确认
  if("1" == m.getAttr("orderId")) {
      //一般这里,需要查询数据库之类的
      m.acknowledge(true);
  }
});
    
//发送事务消息    
const tran = client.newTransaction();

try {
    client.publish("demo", FolkMQ.newMessage("demo1").attr("orderId","1").transaction(tran));
    client.publish("demo", FolkMQ.newMessage("demo2").attr("orderId","1").transaction(tran));
    client.publish("demo", FolkMQ.newMessage("demo3").attr("orderId","1").transaction(tran));
    client.publish("demo", FolkMQ.newMessage("demo4").attr("orderId","1").transaction(tran));

    tran.commit();
} catch (err) {
    tran.rollback();
}
  • 发送与监听模式模式 - rpc for js:
//客户端1
const client1 = await FolkMQ.createClient("folkmq:ws://127.0.0.1:18602")
                        .nameAs("demoapp1")
                        .connect();

//客户端1监听
const router = FolkMQ.newRouter(m => m.getTag()).doOn("/hello", req => {
    req.respone(SocketD.newEntity("me to!"));
});
client1.listen(router.consume.bind(router));


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

//客户端2发送
const resp = client2.send(FolkMQ.newMessage("helloworld!").tag("hello"), "demoapp1").await();
console.log(resp.dataAsString());

二、兼容说明

  • 本次更新,向下兼容。
  • 新增的功能功能(事务消息,发送与监听模式模式,消息属性),需要新版服务端和客户端支持。

FolkMQ v1.2.2

27 Feb 06:05
Compare
Choose a tag to compare

更新说明(优化 rpc 接口体验;强化 trans 概念)

  • 调整 response 拆分为:transactionCheckback 和 listen
  • 调整 request 改为 send
  • 添加 transactionCheckback 用于响应服务端的事务回查
  • 添加 listen 和 send 配套接口
  • 添加 后台图标
  • 完善 许可证本地处理机制
  • sokcet.d 升为 2.4.4

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

//准备(1.取名字;2.添加事务回查)
MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18602")
    .nameAs("demoapp") //一般用当前应用名
    .connect();

//用于服务端发起的事务回查
client.transactionCheckback(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();
}

新功能示例(发送与监听模式模式 - 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());

兼容说明

  • 本次更新,向下兼容。
  • 新增的功能功能(事务消息,发送与监听模式模式,消息属性),需要新版服务端和客户端支持。

FolkMQ v1.2.1

23 Feb 08:07
Compare
Choose a tag to compare

更新说明

  • 添加 协议版本的握手传递
  • 添加 消息事务支持(即二段式提交),支持反向事务确认
  • 添加 请求响应模式支持(即 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());

兼容说明

  • 本次更新,向下兼容。
  • 新增的功能功能(事务消息,RPC模式,消息属性),需要新版服务端和客户端支持。

FolkMQ v1.2.0

23 Feb 07:16
Compare
Choose a tag to compare

更新说明

  • 添加 协议版本的握手传递
  • 添加 消息事务支持(即二段式提交),支持反向事务确认
  • 添加 请求响应模式支持(即 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());

兼容说明

  • 本次更新,向下兼容。
  • 新增的功能功能(事务消息,RPC模式,消息属性),需要新版服务端和客户端支持。

FolkMQ v1.1.0

08 Feb 04:27
Compare
Choose a tag to compare

更新说明

  • 调整 消息流处理改为单线程架构!
  • 添加 “绝对顺序”消息支持(同时支持单机模式与集群模式)!
  • 修复 客户端问题:同一个项目内,用不同的 consumerGroup 订阅同一个topic 会被覆盖的问题
  • sokcet.d 升为 2.4.0

新功能示例:

//同步
client.publish("demo", new MqMessage("hello").sequence(true)));
//异步
client.publishAsync("demo", new MqMessage("hello").sequence(true)));

兼容说明

本次更新,向下兼容。新增的顺序消息功能,需要新版服务端支持。

FolkMQ v1.0.32

06 Feb 02:16
Compare
Choose a tag to compare

更新说明:

  • 服务端的消息主处理,改为单线程模式

向 redis 学习。消息的主处理,基本在内存里进行。所以改成单线程后,性能没啥变化。

  • 添加 单机模式下绝对有序支持
//消息绝对有序配置支持(发送有序,接收有序)
MqClient client = FolkMQ.createClient("folkmq://127.0.0.1:18601?ak=ak1&sk=sk1",
                "folkmq://127.0.0.1:18602?ak=ak1&sk=sk1")
        .config(c->c.sequenceMode(true).coreThreads(1).maxThreads(1))
        .connect();
  • 添加 集群安全停止支持。实现一端停止,另一端完全无感知(不异常,不卡顿)

新的高可用演示视频:https://www.bilibili.com/video/BV1Ha4y1R73b/

  • 添加 管理接口支持
//管理接口,一般用于用户自己开发管理界面,通过接口管理 FolkMQ
String json = client.call(MqApis.MQ_QUEUE_VIEW_MESSAGE, token, "demo","demoApp").get();

兼容说明:

本次更新,向下兼容

FolkMQ v1.0.31

23 Jan 09:39
Compare
Choose a tag to compare

更新说明:

  • sokcet.d 升为 2.3.8

兼容说明:

  • 本次更新,向下兼容

FolkMQ v1.0.30

17 Jan 13:48
Compare
Choose a tag to compare

更新说明:

  • 修复 消息过期判断(修复 1.0.29 的bug)
let msg = new MqMessage("hello").expiration(new Date(System.currentTimeMillis() + 5000));
client.publish("demo", msg);

兼容说明:

  • 本次更新,向下兼容

FolkMQ v1.0.29

16 Jan 08:41
Compare
Choose a tag to compare

更新说明:

  • 添加 消息过期支持
  • sokcet.d 升为 2.3.6
  • solon 升为 2.6.5
let msg = new MqMessage("hello").expiration(new Date(System.currentTimeMillis() + 5000));
client.publish("demo", msg);

兼容说明:

  • 本次更新,向下兼容