在线时间:8:00-16:00
迪恩网络APP
随时随地掌握行业动态
扫描二维码
关注迪恩网络微信公众号
开源软件名称:open-delay-queue开源软件地址:https://gitee.com/corn-cob/open-delay-queue开源软件介绍:DelayQueue延迟消息队列介绍架构: SpringCloud-Stream + RocketMQ + RocksDB(可以使用Nacos,Eureka作为注册中心,作为单独的延迟消息服务使用) 基于Apache RocketMQ的一种延迟消息队列。 由于Apache RocketMQ不支持自定义延迟消息队列,为了支持延时消息队列特开此项目进行对Apache RocketMQ进行补充。以至于达到可以实现自定义延迟的一种方式。 相关Apache RocketMQ消息队列: TOPIC_SYSTEM_QUEUE_REALTIME: 实时消息队列(当消息到期后,此队列消息增加1个) TOPIC_SYSTEM_QUEUE_DELAY: 延迟队列(用于接收延迟消息请求) 1.0.0支持功能项:1、支持水平扩展DelayServer 2、支持自定义延迟消息,可以精确到毫秒级。 请求过程:1、应用调用请求 2、发起延时消息 3、应用方将消息写入 TOPIC_SYSTEM_QUEUE_DELAY 4、延时服务方监听TOPIC_SYSTEM_QUEUE_DELAY 将消息读取出来写入RocksDB 5、延时服务方循环遍历RocksDB检测延时消息是否需要发送 6、如果需要发送则将消息放入TOPIC_SYSTEM_QUEUE_REALTIME 7、应用监听TOPIC_SYSTEM_QUEUE_REALTIME 进行业务操作。 如何扩展?完全沿袭Apache RocketMQ高可用、高性能、低延迟、高可靠的优点,如果RocketMQ性能不够,则增加RocketMQ Broker节点。 如果是消息消费性能不够,则将此服务复制一份,消费组配置成TOPIC_SYSTEM_QUEUE_DELAY 同一个即可。 当然可以改造成Eureka服务,或者Nacos服务,使其Restful也支持高可用。 如何调用提供2种模式,一种是Restful接口,一种是SpringCloudStream Restful接口模式:非延迟消息:GET 1.1 http://delay-server:port/message/push/normal 参数: id: 消息ID(唯一) message: 消息内容 例子: GET 1.1 http://delay-server:port/message/push/normal?id=1&message=Hello World 延迟消息(时间戳):GET 1.1 http://delay-server:port/message/push/delay 参数: id: 消息ID(唯一) message: 消息内容 timestamp:发送时间 例子: GET 1.1 http://delay-server:port/message/push/delay?id=1&message=Hello World×tamp=未来发出消息的时间戳 延迟消息字(符串日期):GET 1.1 http://delay-server:port/message/push/delay/pattern 参数: id: 消息ID(唯一) message: 消息内容 date:字符串日期 pattern:格式化规则 例子: GET 1.1 http://delay-server:port/message/push/delay/pattern?id=1&message=Hello World&date=2021-03-03 11:00:00&pattern=yyyy-MM-dd HH:mm:ss SpringCloudStream模式配置消息生产通道Topic, Topic TOPIC_SYSTEM_QUEUE_DELAY spring: cloud: stream: rocketmq: binder: #新版使用该地址 name-server: rocketmq-server:9876 bindings: #发送通道 system_topic_queue_realtime_output: #目的地:主题(Topic) destination: TOPIC_SYSTEM_QUEUE_REALTIME #发送通道 system_topic_queue_delay_output: #目的地:主题(Topic) destination: TOPIC_SYSTEM_QUEUE_DELAY #接收通道 system_topic_queue_delay_input: #目的地:主题(Topic) destination: TOPIC_SYSTEM_QUEUE_DELAY #MessageConvert 类型, JSON协议 content-type: application/json #消费组 group: TOPIC_SYSTEM_QUEUE_DELAY_GROUP application: name: omuao-delay-queue jackson: date-format: yyyy-MM-dd HH:mm:ss time-zone: GMT+8 default-property-inclusion: non_null mvc: date-format: yyyy-MM-dd HH:mm:ss servlet: multipart: max-file-size: 10MB max-request-size: 10MB |
请发表评论