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

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"
  }
}

@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)
      }
    }
  }
}

@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
  }
}

@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
  }
}