如何根据用户定义的参数执行节流？

Question

我正在写入一个内存分布式数据库，批量大小是用户在多线程环境中定义的。但我想限制写入 ex 的行数。 1000 行/秒。这个要求的原因是我的生产者写得太快了，而消费者遇到了叶子内存错误。在批处理记录时是否有任何标准做法来执行节流。

dataStream.map(line => readJsonFromString(line)).grouped(memsqlBatchSize).foreach { recordSet =>
      val dbRecords = recordSet.map(m => (m, Events.transform(m)))
      dbRecords.map { record =>
        try {
          Events.setValues(eventInsert, record._2)
          eventInsert.addBatch
        } catch {
          case e: Exception =>
            logger.error(s"error adding batch: ${e.getMessage}")
            val error_event = Events.jm.writeValueAsString(mapAsJavaMap(record._1.asInstanceOf[Map[String, Object]]))
            logger.error(s"event: $error_event")
        }
      }

      // Bulk Commit Records
      try {
        eventInsert.executeBatch
      } catch {
        case e: java.sql.BatchUpdateException =>
          val updates = e.getUpdateCounts
          logger.error(s"failed commit: ${updates.toString}")
          updates.zipWithIndex.filter { case (v, i) => v == Statement.EXECUTE_FAILED }.foreach { case (v, i) =>
            val error = Events.jm.writeValueAsString(mapAsJavaMap(dbRecords(i)._1.asInstanceOf[Map[String, Object]]))
            logger.error(s"insert error: $error")
            logger.error(e.getMessage)
          }
      }
      finally {
        connection.commit
        eventInsert.clearBatch
        logger.debug(s"committed: ${dbRecords.length.toString}")
      }
    }

我希望如果我可以将用户定义的参数作为throttleMax 传递，并且如果每个线程写入的总记录达到throttleMax，thread.sleep() 将被调用1 秒。但这会使整个过程变得非常缓慢。有没有其他有效的方法可以将数据的加载速度限制在 1000 行/秒？

Answer 1

正如其他人所建议的那样（请参阅问题中的 cmets），您可以选择比在这里节流更好的选择。但是，您可以使用一些简单的代码来限制 Java 中的操作，如下所示：

/**
 * Given an Iterator `inner`, returns a new Iterator which will emit items upon
 * request, but throttled to at most one item every `minDelayMs` milliseconds.
 */
public static <T> Iterator<T> throttledIterator(Iterator<T> inner, int minDelayMs) {
    return new Iterator<T>() {
        private long lastEmittedMillis = System.currentTimeMillis() - minDelayMs;

        @Override
        public boolean hasNext() {
            return inner.hasNext();
        }

        @Override
        public T next() {
            long now = System.currentTimeMillis();
            long requiredDelayMs = now - lastEmittedMillis;
            if (requiredDelayMs > 0) {
                try {
                    Thread.sleep(requiredDelayMs);
                } catch (InterruptedException e) {
                    // resume
                }
            }
            lastEmittedMillis = now;

            return inner.next();
        }
    };
}

上面的代码使用Thread.sleep，所以不适合在Reactive系统中使用。在这种情况下，您可能希望使用该系统中提供的 Throttle 实现，例如throttle in Akka