使用Kotlin+RocketMQ实现延时消息的示例代码

这篇文章主要介绍了使用Kotlin+RocketMQ实现延时消息的示例代码,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

一. 延时消息

延时消息是指消息被发送以后,并不想让消费者立即拿到消息,而是等待指定时间后,消费者才拿到这个消息进行消费。

使用延时消息的典型场景,例如:

  • 在电商系统中,用户下完订单30分钟内没支付,则订单可能会被取消。
  • 在电商系统中,用户七天内没有评价商品,则默认好评。

这些场景对应的解决方案,包括:

  • 轮询遍历数据库记录
  • JDK 的 DelayQueue
  • ScheduledExecutorService
  • 基于 Quartz 的定时任务
  • 基于 Redis 的 zset 实现延时队列。

除此之外,还可以使用消息队列来实现延时消息,例如 RocketMQ。

二. RocketMQ

RocketMQ 是一个分布式消息和流数据平台,具有低延迟、高性能、高可靠性、万亿级容量和灵活的可扩展性。RocketMQ 是2012年阿里巴巴开源的第三代分布式消息中间件。

三. RocketMQ 实现延时消息

3.1 业务背景

我们的系统完成某项操作之后,会推送事件消息到业务方的接口。当我们调用业务方的通知接口返回值为成功时,表示本次推送消息成功;当返回值为失败时,则会多次推送消息,直到返回成功为止(保证至少成功一次)。
当我们推送失败后,虽然会进行多次推送消息,但并不是立即进行。会有一定的延迟,并按照一定的规则进行推送消息。
例如:1小时后尝试推送、3小时后尝试推送、1天后尝试推送、3天后尝试推送等等。因此,考虑使用延时消息实现该功能。

3.2 生产者(Producer)

生产者负责产生消息,生产者向消息服务器发送由业务应用程序系统生成的消息。

首先,定义一个支持延时发送的 AbstractProducer。


abstract class AbstractProducer :ProducerBean() {
  var producerId: String? = null
  var topic: String? = null
  var tag: String?=null
  var timeoutMillis: Int? = null
  var delaySendTimeMills: Long? = null

  val log = LogFactory.getLog(this.javaClass)

  open fun sendMessage(messageBody: Any, tag: String) {
    val msgBody = JSON.toJSONString(messageBody)
    val message = Message(topic, tag, msgBody.toByteArray())

    if (delaySendTimeMills != null) {
      val startDeliverTime = System.currentTimeMillis() + delaySendTimeMills!!
      message.startDeliverTime = startDeliverTime
      log.info( "send delay message producer startDeliverTime:${startDeliverTime}currentTime :${System.currentTimeMillis()}")
    }
    val logMessageId = buildLogMessageId(message)
    try {
      val sendResult = send(message)
      log.info(logMessageId + "producer messageId: " + sendResult.getMessageId() + "\n" + "messageBody: " + msgBody)
    } catch (e: Exception) {
      log.error(logMessageId + "messageBody: " + msgBody + "\n" + " error: " + e.message, e)
    }

  }

  fun buildLogMessageId(message: Message): String {
    return "topic: " + message.topic + "\n" +
        "producer: " + producerId + "\n" +
        "tag: " + message.tag + "\n" +
        "key: " + message.key + "\n"
  }
}

根据业务需要,增加一个支持重试机制的 Producer


@Component
@ConfigurationProperties("mqs.ons.producers.xxx-producer")
@Configuration
@Data
class CleanReportPushEventProducer :AbstractProducer() {

  lateinit var delaySecondList:List<Long>

  fun sendMessage(messageBody: CleanReportPushEventMessage){
    //重试超过次数之后不再发事件
    if (delaySecondList!=null) {

      if(messageBody.times>=delaySecondList.size){
        return
      }
      val msgBody = JSON.toJSONString(messageBody)
      val message = Message(topic, tag, msgBody.toByteArray())
      val delayTimeMills = delaySecondList[messageBody.times]*1000L
      message.startDeliverTime = System.currentTimeMillis() + delayTimeMills
      log.info( "messageBody: " + msgBody+ "startDeliverTime: "+message.startDeliverTime )
      val logMessageId = buildLogMessageId(message)
      try {
        val sendResult = send(message)
        log.info(logMessageId + "producer messageId: " + sendResult.getMessageId() + "\n" + "messageBody: " + msgBody)
      } catch (e: Exception) {
        log.error(logMessageId + "messageBody: " + msgBody + "\n" + " error: " + e.message, e)
      }
    }
  }
}

在 CleanReportPushEventProducer 中,超过了重试的次数就不会再发送消息了。

每一次延时消息的时间也会不同,因此需要根据重试的次数来获取这个delayTimeMills 。

通过 System.currentTimeMillis() + delayTimeMills 可以设置 message 的 startDeliverTime。然后调用 send(message) 即可发送延时消息。

我们使用商用版的 RocketMQ,因此支持精度为秒级别的延迟消息。在开源版本中,RocketMQ 只支持18个特定级别的延迟消息。:(

3.3 消费者(Consumer)

消费者负责消费消息,消费者从消息服务器拉取信息并将其输入用户应用程序。

定义 Push 类型的 AbstractConsumer:


@Data
abstract class AbstractConsumer ():MessageListener{

  var consumerId: String? = null

  lateinit var subscribeOptions: List<SubscribeOptions>

  var threadNums: Int? = null

  val log = LogFactory.getLog(this.javaClass)

  override fun consume(message: Message, context: ConsumeContext): Action {
    val logMessageId = buildLogMessageId(message)
    val body = String(message.body)
    try {
      log.info(logMessageId + " body: " + body)
      val result = consumeInternal(message, context, JSON.parseObject(body, getMessageBodyType(message.tag)))
      log.info(logMessageId + " result: " + result.name)
      return result
    } catch (e: Exception) {
      if (message.reconsumeTimes >= 3) {
        log.error(logMessageId + " error: " + e.message, e)
      }
      return Action.ReconsumeLater
    }

  }

  abstract fun getMessageBodyType(tag: String): Type?

  abstract fun consumeInternal(message: Message, context: ConsumeContext, obj: Any): Action

  protected fun buildLogMessageId(message: Message): String {
    return "topic: " + message.topic + "\n" +
        "consumer: " + consumerId + "\n" +
        "tag: " + message.tag + "\n" +
        "key: " + message.key + "\n" +
        "MsgId:" + message.msgID + "\n" +
        "BornTimestamp" + message.bornTimestamp + "\n" +
        "StartDeliverTime:" + message.startDeliverTime + "\n" +
        "ReconsumeTimes:" + message.reconsumeTimes + "\n"
  }
}

再定义具体的消费者,并且在消费失败之后能够再发送一次消息。


@Configuration
@ConfigurationProperties("mqs.ons.consumers.clean-report-push-event-consumer")
@Data
class CleanReportPushEventConsumer(val cleanReportService: CleanReportService,val eventProducer:CleanReportPushEventProducer):AbstractConsumer() {

  val logger: Logger = LoggerFactory.getLogger(this.javaClass)

  override fun consumeInternal(message: Message, context: ConsumeContext, obj: Any): Action {
    if(obj is CleanReportPushEventMessage){
      //清除事件
      logger.info("consumer clean-report event report_id:${obj.id} ")

      //消费失败之后再发送一次消息
      if(!cleanReportService.sendCleanReportEvent(obj.id)){
        val times = obj.times+1
        eventProducer.sendMessage(CleanReportPushEventMessage(obj.id,times))
      }
    }
    return Action.CommitMessage
  }

  override fun getMessageBodyType(tag: String): Type? {
    return CleanReportPushEventMessage::class.java
  }
}

其中,cleanReportService 的 sendCleanReportEvent() 会通过 http 的方式调用业务方提供的接口,进行事件消息的推送。如果推送失败了,则会进行下一次的推送。(这里使用了 eventProducer 的 sendMessage() 方法再次投递消息,是因为要根据调用的http接口返回的内容来判断消息是否发送成功。)

最后,定义 ConsumerFactory


@Component
class ConsumerFactory(val consumers: List<AbstractConsumer>,val aliyunOnsOptions: AliyunOnsOptions) {

  val logger: Logger = LoggerFactory.getLogger(this.javaClass)


  @PostConstruct
  fun start() {
    CompletableFuture.runAsync{
      consumers.stream().forEach {
        val properties = buildProperties(it.consumerId!!, it.threadNums)
        val consumer = ONSFactory.createConsumer(properties)
        if (it.subscribeOptions != null && !it.subscribeOptions!!.isEmpty()) {
          for (options in it.subscribeOptions!!) {
            consumer.subscribe(options.topic, options.tag, it)
          }
          consumer.start()
          val message = "\n".plus(
              it.subscribeOptions!!.stream().map{ a -> String.format("topic: %s, tag: %s has been started", a.topic, a.tag)}
                  .collect(Collectors.toList<Any>()))
          logger.info(String.format("consumer: %s\n", message))
        }
      }
    }
  }

  private fun buildProperties(consumerId: String,threadNums: Int?): Properties {
    val properties = Properties()
    properties.put(PropertyKeyConst.ConsumerId, consumerId)
    properties.put(PropertyKeyConst.AccessKey, aliyunOnsOptions.accessKey)
    properties.put(PropertyKeyConst.SecretKey, aliyunOnsOptions.secretKey)
    if (StringUtils.isNotEmpty(aliyunOnsOptions.onsAddr)) {
      properties.put(PropertyKeyConst.ONSAddr, aliyunOnsOptions.onsAddr)
    } else {
      // 测试环境接入RocketMQ
      properties.put(PropertyKeyConst.NAMESRV_ADDR, aliyunOnsOptions.nameServerAddress)
    }
    properties.put(PropertyKeyConst.ConsumeThreadNums, threadNums!!)
    return properties
  }
}

四. 总结

正如本文开头曾介绍过,可以使用多种方式来实现延时消息。然而,我们的系统本身就大量使用了 RocketMQ,借助成熟的 RocketMQ 实现延时消息不失为一种可靠而又方便的方式。

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

本站部分内容来源互联网,如果有图片或者内容侵犯您的权益请联系我们删除!

相关文档推荐

这篇文章主要给大家介绍了关于Kotlin函数默认值的相关资料,文中通过示例代码介绍的非常详细,对大家学习或者使用Kotlin具有一定的参考学习价值,需要的朋友们下面来一起学习学习吧
这篇文章主要给大家介绍了关于kotlin实战教程之lambda编程的相关资料,文中通过示例代码介绍的非常详细,对大家学习或者使用kotlin具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
这篇文章主要给大家介绍了关于Kotlin如何优雅地判断EditText数据是否为空的相关资料,文中通过示例代码介绍的非常详细,对大家学习或者使用kotlin具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧
这篇文章主要介绍了Android Kotlin仿微信头像裁剪图片的方法示例,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧
Android自定义View实现简单水波纹效果 本文实例为大家分享了Android自定义View实现水波纹效果的具体代码,供大家参考,具体内容如下 效果如下: 原理 控制代码 //这里用的kotlin //主线程刷新控件  val mHandler = object : Handler() {         override fun handleMes
Android实现倒计时的方案梳理 目录 前言 一.CountDownTimer的实现 二.直接用Handler的实现 三.直接用Time.TimeTask的实现 四.使用Theard倒计时 五.使用框架RxJava 六.Kotlin Flow 的实现 总结 前言 关于倒计时可以说我们App开发中常见的一种场景了,比如Splash倒计时