Будущий обратный вызов onComplete не вызывается

Question

Я пытаюсь записать истекшее время каждой задачи следующим образом.

Проблема, с которой я столкнулся, заключается в том, что метод обратного вызова в logElapsedTime никогда не вызывается по какой-либо причине.

Вызывается только самый последний обратный вызов в будущее f .

Как я могу исправить это, чтобы каждое прошедшее время было правильно зарегистрировано?

    def logElapsedTime[T](f: Future[T], description: String): Future[T] = {
      val start = System.currentTimeMillis()
      f onComplete (_ => logger.debug(s"$description took [${System.currentTimeMillis() - start}]"))
      f
    }

    val f = for {
      _ <- logElapsedTime(task1(), "1st task to be executed")
      result <- logElapsedTime(task2(), "2nd task to be executed")
      _ <- logElapsedTime(task3(), "3rd task to be executed")
      _ <- logElapsedTime(task4(), "4th task to be executed")
    } yield result

    f onComplete {
      case Success(v) =>
        logger.info(s"tasks succeeded !!!! $v")
      case Failure(ex) =>
        logger.error(ex.getMessage)
        throw ex
    }   

Примеры выходных данных ↓

при успешном выполнении:

задание выполнено успешно !!!! некоторое значение

при ошибке:

некоторое сообщение об ошибке

Другие входы не регистрируются.
(Уровень журнала установлен на отладку и выше)

Answer 1

Рассмотрим andThen, когда мы просто хотим выполнить регистрацию как побочный эффект без преобразования значения внутри Future, например

object futureAndThenLogging extends App with LazyLogging {

  def logElapsedTime[T](f: Future[T], description: String): Future[T] = {
    val start = System.currentTimeMillis()
    f andThen { case _ => logger.debug(s"$description took [${System.currentTimeMillis() - start}]") }
  }

  def task1() = Future(1)
  def task2() = Future(2)
  def task3() = Future(3)
  def task4() = Future(4)

  (for {
    _ <- logElapsedTime(task1(), "1st task to be executed")
    result <- logElapsedTime(task2(), "2nd task to be executed")
    _ <- logElapsedTime(task3(), "3rd task to be executed")
    _ <- logElapsedTime(task4(), "4th task to be executed")
  } yield result)
    .andThen {
      case Success(v) => logger.info(s"tasks succeeded !!!! $v")
      case Failure(ex) => logger.error(ex.getMessage)
    }

  Thread.sleep(1000) // just for demonstration purposes
}

Обратите внимание, что мы этого не сделалидолжны повторно throw ex в case Failure(ex) => logger.error(ex.getMessage).

Answer 2

В вашей логике нет ничего плохого. Применено несколько модификаций, которые я предлагаю вам попробовать.

import org.slf4j.LoggerFactory

import scala.concurrent.Future
import concurrent.ExecutionContext.Implicits.global
import scala.io.StdIn
import scala.util.{Failure, Success}

object FutureOnComplete extends App {

  private val logger = LoggerFactory.getLogger("test")

  def logElapsedTime[T](f: => Future[T], description: String): Future[T] = {
    val start = System.currentTimeMillis()
    f.onComplete(
      _ =>
        logger.warn(
          s"$description took [${System.currentTimeMillis() - start}]"))
    f
  }

  val f = for {
    _ <- logElapsedTime(Future(1), "1st task to be executed")
    result <- logElapsedTime(Future(2), "2nd task to be executed")
    _ <- logElapsedTime(Future(2), "3rd task to be executed")
    _ <- logElapsedTime(Future(2), "4th task to be executed")
  } yield result

  f.onComplete {
    case Success(v) =>
      logger.info(s"tasks succeeded !!!! $v")
    case Failure(ex) =>
      logger.error(ex.getMessage)
      throw ex
  }

  StdIn.readLine()

}

• Увеличьте уровень журнала до warn, чтобы убедиться, что ваша регистрация не виновата. Или замените его на println
• Подождите в вашем основном потоке для будущего завершения, например, StdIn.readLine(). Это позволяет завершить асинхронные процессы и запустить onComplete.
• Использовать => Future[T] по имени параметра, чтобы начать выполнение будущего внутри метода logElapsedTime. Это только изменяется, когда будущее начинается, но не логика регистрации