我正在开发一个系统，从jms（消费者）中提取消息并将其推送到kafka主题（生产者）。
既然我的消费者保持活动状态，等待新消息到达jms队列并将其推送到kafka，那么我如何有效地度量每秒可以拉取多少消息呢？
这是我的密码：
我的消费者：

class ActiveMqConsumerActor extends Consumer {

  var startTime: Long = _

  val log = Logging(context.system, this)

  val producerActor = context.actorOf(Props[KafkaProducerActor])

  override def autoAck = false
  override def endpointUri: String = "activemq:KafkaTest"

  override def receive: Receive = LoggingReceive {
    case msg: CamelMessage =>
      val camelMsg = msg.bodyAs[String]
      producerActor ! Message(camelMsg.getBytes)
      sender() ! Ack

    case ex: Exception => sender() ! Failure(ex)

    case _ =>
      log.error("Got a message that I don't understand")
      sender() ! Failure(new Exception("Got a message that I don't understand"))
  }
}

主要内容：

object ActiveMqConsumerTest extends App {

  val system = ActorSystem("KafkaSystem")
  val camel = CamelExtension(system)
  val camelContext = camel.context

  camelContext.addComponent("activemq", ActiveMQComponent.activeMQComponent("tcp://0.0.0.0:61616"))

  val consumer = system.actorOf(Props[ActiveMqConsumerActor].withRouter(FromConfig), "consumer")
  val producer = system.actorOf(Props[KafkaProducerActor].withRouter(FromConfig), "producer")

}

谢谢