golang实现redis的延时消息队列功能示例

前言

在学习过程中发现redis的zset还可以用来实现轻量级的延时消息队列功能,虽然可靠性还有待提高,但是对于一些对数据可靠性要求不那么高的功能要求完全可以实现。本次主要采用了redis中zset中的zadd, zrangebyscore 和 zdel来实现一个小demo。

提前准备 安装redis, redis-go

因为用的是macOS, 直接

$ brew install redis
$ go get github.com/garyburd/redigo/redis

又因为比较懒,生成任务的唯一id时,直接采用了bson中的objectId,所以:

$ go get gopkg.in/mgo.v2/bson

唯一id不是必须有,但如果之后有实际应用需要携带,便于查找相应任务。

生产者

通过一个for循环生成10w个任务, 每一个任务有不同的时间

func producer() {
 count := 0
 //生成100000个任务
 for count < 100000 {
 count++
 dealTime := int64(rand.Intn(5)) + time.Now().Unix()
 uuid := bson.NewObjectId().Hex()
 redis.Client.AddJob(&job.JobMessage{
 Id: uuid,
 DealTime: dealTime,
 }, + int64(dealTime))
 }
}

其中AddJob函数在另一个包中, 将上一个函数中随机生成的时间作为需要处理的时间戳.

// 添加任务
func (client *RedisClient) AddJob(msg *job.JobMessage, dealTime int64) {
 conn := client.Get()
 defer conn.Close()

 key := "JOB_MESSAGE_QUEUE"
 conn.Do("zadd", key, dealTime, util.JsonEncode(msg))
}

消费者

消费者处理流程分为两个步骤:

  • 获取小于等于当前时间戳的任务
  • 通过删除当前任务来判断谁获得了当前任务

因为在获取小于等于当前时间戳的任务时,可能有多个go routine同时读到了当前任务,而只有一个任务可以来处理当前任务。因此我们需要通过一个方案来判断究竟由谁来处理这个任务(当然如果只有一个消费者可以读到就直接处理):这个时候可以通过redis的删除操作来获取,因为删除指定value时只有成功的操作才会返回不为0,所以我们可以认为删除当前队列成功的那个go routine拿到了当前的任务。

下面是代码:

// 消费者
func consumer() {
 // 启动10个go routine一起去拿
 count := 0
 for count < 10 {
 go func() {
 for {
 jobs := redis.Client.GetJob()
 if len(jobs) <= 0 {
  time.Sleep(time.Second * 1)
  continue
 }
 currentJob := jobs[0]
 // 如果当前抢redis队列成功,
 if redis.Client.DelJob(currentJob) > 0 {
  var jobMessage job.JobMessage
  util.JsonDecode(currentJob, &jobMessage) //自定义的json解析函数
  handleMessage(&jobMessage)
 }

 }

 }()
 count++
 }
}

// 处理任务用函数
func handleMessage(msg *job.JobMessage) {
 fmt.Printf("deal job: %s, require time: %d \n", msg.Id, msg.DealTime)
 go func() {
 countChan <- true
 }()
}

redis部分的代码,获取任务和删除任务

// 获取任务
func (client *RedisClient) GetJob() []string {
 conn := client.Get()
 defer conn.Close()

 key := "JOB_MESSAGE_QUEUE"
 timeNow := time.Now().Unix()
 ret, err := redis.Strings(conn.Do("zrangebyscore", key, 0, timeNow, "limit", 0, 1))
 if err != nil {
 panic(err)
 }
 return ret
}

// 删除当前任务, 用来判断是否抢到了当前任务
func (client *RedisClient) DelJob(value string) int {
 conn := client.Get()
 defer conn.Close()

 key := "JOB_MESSAGE_QUEUE"
 ret, err := redis.Int(conn.Do("zrem", key, value))
 if err != nil {
 panic(err)
 }
 return ret
}

代码大抵如此。最后跑起来之后,大概每3-4秒钟能够处理掉1w个任务,速度上确实是...

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

(0)

相关推荐

  • 在Golang中使用Redis的方法示例

    周五上班的主要任务是在公司老平台上用redis处理一个队列问题,顺便复习了一下redis操作的基础知识,回来后就想着在自己的博客demo里,用redis来优化一些使用场景,学习一下golang开发下redis的使用. Redis简单介绍 简介 关于Redis的讨论,其实在现在的后台开发中已经是个老生常谈的问题,基本上也是后端开发面试的基本考察点.其中 Redis的背景介绍和细节说明在这里就不赘述.不管怎么介绍,核心在于Redis是一个基于内存的key-value的多数据结构存储,并可以提供持久化

  • 用golang实现一个定时器任务队列实例

    很有幸得到公司信任,采用新的语言进行一些底层服务的开发,在实现功能的同时,也获得了一些感悟,因此在这记录一下,方便自己查看也可以共享给大家. golang中定时器 golang中提供了2种定时器timer和ticker(如果JS很熟悉的话应该会很了解),分别是一次性定时器和重复任务定时器. 一般用法: func main() { input := make(chan interface{}) //producer - produce the messages go func() { for i

  • 使用golang写一个redis-cli的方法示例

    0. redis通信协议 redis的客户端(redis-cli)和服务端(redis-server)的通信是建立在tcp连接之上, 两者之间数据传输的编码解码方式就是所谓的redis通信协议.所以,只要我们的redis-cli实现了这个协议的解析和编码,那么我们就可以完成所有的redis操作. redis 协议设计的非常易读,也易于实现,关于具体的redis通信协议请参考:通信协议(protocol).后面我们在实现这个协议的过程中也会简单重复介绍一下具体实现 1. 建立tcp连接 redis

  • golang环形队列实现代码示例

    Summary 什么是环形队列 实现环形队列图示过程 golang版本代码实现过程 参考全部代码 什么是环形队列 在一个指定大小的数组里循环写入数据,借用二个指针分别实现入队标记与出队标记.也体现了指针的大好用处,请深入体会.大有裨益. 如图所示,一个环形队列.含有二个指针: 队列头指针,队列尾指针. 实现环形队列图示过程 初始化一个数组大小为6的环形队列, 头指针front=0, 尾指针rear=0, 刚好front=rear =0的状态,表示环形队列为空. 2.向环形队列里插入1个元素,则r

  • golang实现redis的延时消息队列功能示例

    前言 在学习过程中发现redis的zset还可以用来实现轻量级的延时消息队列功能,虽然可靠性还有待提高,但是对于一些对数据可靠性要求不那么高的功能要求完全可以实现.本次主要采用了redis中zset中的zadd, zrangebyscore 和 zdel来实现一个小demo. 提前准备 安装redis, redis-go 因为用的是macOS, 直接 $ brew install redis $ go get github.com/garyburd/redigo/redis 又因为比较懒,生成任

  • php+redis实现消息队列功能示例

    本文实例讲述了php+redis实现消息队列功能.分享给大家供大家参考,具体如下: 个人理解在项目中使用消息队列一般是有如下几个原因: 把瞬间服务器的请求处理换成异步处理,缓解服务器的压力 实现数据顺序排列获取 redis实现消息队列步骤如下: 1).redis函数rpush,lpop 2).建议定时任务入队列 3)创建定时任务出队列 文件:demo.php插入数据到redis队列 <?php $redis = new Redis(); $redis->connect('127.0.0.1',

  • Java实现Redis延时消息队列

    目录 什么是延时任务 延时任务的特点 实现思路: 代码实现 1.消息模型 2.RedisMq 消息队列实现类 3.消息生产者 4.消息消费者 5. 消息执接口 6. 任务类型的实现类:可以根据自己的情况去实现对应的队列需求 什么是延时任务 延时任务,顾名思义,就是延迟一段时间后才执行的任务.举个例子,假设我们有个发布资讯的功能,运营需要在每天早上7点准时发布资讯,但是早上7点大家都还没上班,这个时候就可以使用延时任务来实现资讯的延时发布了.只要在前一天下班前指定第二天要发送资讯的时间,到了第二天

  • 使用 Redis 流实现消息队列的代码

    在介绍了 Redis 流的基本功能之后, 现在是时候使用这些功能来构建一些实际的应用了. 消息队列作为流的典型应用之一, 具有非常好的示范性, 因此我们将使用 Redis 流的相关功能构建一个消息队列应用, 这个消息队列跟我们之前使用其他 Redis 数据结构构建的消息队列具有相似的功能. 代码清单 10-1 展示了一个具有基本功能的消息队列实现: 代码最开头的是几个转换函数, 它们负责对程序的相关输入输出进行转换和格式化: MessageQueue 类用于实现消息队列, 它的添加消息.移除消息

  • redis stream 实现消息队列的实践

    目录 redis 实现消息对列4中方法 发布订阅 list 队列 zset 队列 Stream 队列 基本命令 xadd 生产消息 读取消息 xgroup 消费者组 xreadgroup 消费消息 Pending 等待列表 消息确认 消息转移 信息监控 SpringBoot 整合 参考文档: Redis5.0带来了Stream类型.从字面上看是流类型,但其实从功能上看,应该是Redis对消息队列(MQ,Message Queue)的完善实现. 基于redis实现消息队列的方式有很多: PUB/S

  • Java利用Redis实现消息队列的示例代码

    本文介绍了Java利用Redis实现消息队列的示例代码,分享给大家,具体如下: 应用场景 为什么要用redis? 二进制存储.java序列化传输.IO连接数高.连接频繁 一.序列化 这里编写了一个java序列化的工具,主要是将对象转化为byte数组,和根据byte数组反序列化成java对象; 主要是用到了ByteArrayOutputStream和ByteArrayInputStream; 注意:每个需要序列化的对象都要实现Serializable接口; 其代码如下: package Utils

  • SpringBoot AOP Redis实现延时双删功能实战

    目录 一.业务场景 1.此时存在的问题 2.解决方案 3.为何要延时500毫秒? 4.为何要两次删除缓存? 二.代码实践 1.引入Redis和SpringBoot AOP依赖 2.编写自定义aop注解和切面 3.application.yml 4.user_db.sql脚本 5.UserController 6.UserService 三.测试验证 四.代码工程及地址 一.业务场景 在多线程并发情况下,假设有两个数据库修改请求,为保证数据库与redis的数据一致性,修改请求的实现中需要修改数据库

  • SpringBoot中使用Redis Stream实现消息监听示例

    目录 Demo环境 仓库地址 POM依赖 配置监听消息类 监听俩个stream的实现 [问题补充]确认完消息删除消息 [问题补充]自动初始化stream的key和group问题-最新更新-2021年12月4日 Demo环境 JDK8 Maven3.6.3 springboot2.4.3 redis_version:6.2.1 仓库地址 https://gitee.com/hlovez/redismq.git. POM依赖 <?xml version="1.0" encoding=

  • 使用go实现一个超级mini的消息队列的示例代码

    目录 前言 目的 设计 协议 队列 broker 删除消息 生产者 消费者 启动 总结 前言 趁着有空余时间,就想着撸一个mini的生产-消费消息队列,说干就干了.自己是个javer,这次实现,特意换用了go.没错,是零基础上手go,顺便可以学学go. 前置知识: go基本语法 消息队列概念,也就三个:生产者.消费者.队列 目的 没想着实现多复杂,因为时间有限,就mini就好,mini到什么程度呢 使用双向链表数据结构作为队列 有多个topic可供生产者生成消息和消费者消费消息 支持生产者并发写

  • Spring boot 整合KAFKA消息队列的示例

    这里使用 spring-kafka 依赖和 KafkaTemplate 对象来操作 Kafka 服务. 一.添加依赖和添加配置项 1.1.在 Pom 文件中添加依赖 <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> </dependency> 1.2.添加配置项 spring: kafka: b

随机推荐