NATS流媒体服务器订户速率限制和精确一次传送

Question

我有点玩NATS流，我有一个问题的用户速率限制。当我将飞行中的最大值设置为1，超时设置为1秒时，我有一个消费者，基本上是一个Thread.sleep(1000)，那么我得到相同事件的多次。我认为，通过限制在飞行和使用手动攻击，这是不应该发生的。一旦交付给速度非常慢的消费者，我怎样才能得到满意的服务？

  case class EventBus[I, O](inputTopic: String, outputTopic: String, connection: Connection, eventProcessor: StatefulEventProcessor[I, O]) {
    // the event bus could be some abstract class while the `Connection` coulbd be injected using DI
    val substritionOptions: SubscriptionOptions = new SubscriptionOptions.Builder()
                                                                         .setManualAcks(true)
                                                                         .setDurableName("foo")
                                                                         .setMaxInFlight(1)
                                                                         .setAckWait(1, TimeUnit.SECONDS)
                                                                         .build()

    if (!inputTopic.isEmpty) {
      connection.subscribe(inputTopic, new MessageHandler() {
        override def onMessage(m: Message) {
          m.ack()
          try {
            val event = eventProcessor.deserialize(m.getData)
            eventProcessor.onEvent(event)
          } catch {
            case any =>
              try {
                val command = new String(m.getData)
                eventProcessor.onCommand(command)
              } catch {
                case any => println(s"de-serialization error: $any")
              }
          } finally {
            println("got event")
          }
        }
      }, substritionOptions)
    }

    if (!outputTopic.isEmpty) {
      eventProcessor.setBus(e => {
        try {
          connection.publish(outputTopic, eventProcessor.serialize(e))
        } catch {
          case ex => println(s"serialization error $ex")
        }
      })
    }
  }


  abstract class StatefulEventProcessor[I, O] {
    private var bus: Option[O => Unit] = None
    def onEvent(event: I): Unit
    def onCommand(command: String): Unit

    def serialize(o: O): Array[Byte] =
      SerializationUtils.serialize(o.asInstanceOf[java.io.Serializable])

    def deserialize(in: Array[Byte]): I =
      SerializationUtils.deserialize[I](in)

    def setBus(push: O => Unit): Unit = {
      if (bus.isDefined) {
        throw new IllegalStateException("bus already set")
      } else {
        bus = Some(push)
      }
    }

    def push(event: O) =
      bus.get.apply(event)
  }


  EventBus("out-1", "out-2", sc, new StatefulEventProcessor[String, String] {
    override def onEvent(event: String): Unit = {
      Thread.sleep(1000)
      push("!!!" + event)
    }

    override def onCommand(command: String): Unit = {}
  })

  (0 until 100).foreach(i => sc.publish("out-1", SerializationUtils.serialize(s"test-$i")))

Answer 1

首先，在NATS流中没有确切的一次(再)交付保证。MaxInflight提供给您的是，在未确认消息的数量低于该数目之前，服务器不会向订阅服务器发送新消息。因此，在MaxInflight(1)的情况下，只有在从先前传递的消息接收到ack之后，才要求服务器发送下一条新消息。但是，这不会阻止未确认消息的重新传递。

服务器无法保证或不知道订户实际接收到消息。这就是ACK的目的，让服务器知道订阅服务器正确地处理了消息。如果服务器不执行重发(即使在到达MaxInflight时)，那么“丢失”消息将永远停止您的订阅。请记住，NATS流服务器和客户端不是通过TCP连接彼此直接连接(它们都连接到NATS服务器，又名gnatsd)。