在 ScheduledExecutorService 中使用 Java stream forEach() 冻结

Question

一般的想法是在后台每 10 秒有一个 Runnable 运行来检查一些数据，并在需要时对对象进行更改。 ScheduledExecutorService 在方法 main() 中实例化，任务被调度。 Runnable 任务实例化 Crawler 对象并开始爬取。大多数情况下，它会成功运行几次，但是当应用程序运行并且数据发生更改时，其中一个爬虫方法会被触发但永远不会结束。代码中没有循环。我试图调试也没有成功。也许你会发现问题所在。

主线：

public class Main {

    public static void main(String[] args) {

        DataStock dataStock = DataStock.getInstance();
        ScheduledExecutorService ses = Executors.newSingleThreadScheduledExecutor();
        ses.scheduleAtFixedRate(new EveryFiveSeconds(), 5, 5, TimeUnit.SECONDS);
        // below the task which fails after couple of runs
        ses.scheduleAtFixedRate(new EveryTenSeconds(), 1 , 10, TimeUnit.SECONDS);
        dataStock.init();

        Menu currentScreen = new UserMenu();
        while(currentScreen != null) {
            currentScreen = currentScreen.display();
        }

    }
}

每十秒可运行：

public class EveryTenSeconds implements Runnable {
    @Override
    public void run() {
        Crawler crawler = new Crawler();
        crawler.crawl();
    }
}

抓取工具：

public class Crawler {

    private final DataStock dataStock;

    public Crawler() {
        this.dataStock = DataStock.getInstance();
    }

    public void crawl() {
        checkOutRentables(dataStock.getCarServicesWithOwners().keySet());
        checkFinancialBook(dataStock.getPaymentsBook(), dataStock.getCurrentDate());
    }

    private void checkOutRentables(Set<CarService> carServices) {
        System.out.println("Start check...");
        carServices.stream()
                .flatMap(service -> service.getWarehousesSet().stream())
                .filter(rentable -> !rentable.isAvailableForRent())
                .forEach(RentableArea::refreshCurrentState);
        System.out.println("Checking finished");
    }

    private void checkFinancialBook(Set<BookEntry> bookEntries, LocalDate currentDate) {
        System.out.println("Start second check...");
        bookEntries.stream()
                .filter(bookEntry -> currentDate.isAfter(bookEntry.getPaymentDeadline()) && !bookEntry.isPaid() && !bookEntry.isNotified())
                .forEach(BookEntry::notifyDebtor);
        System.out.println("Finished second check..."); //this line never shows in one of runs and the task is never repeated again...

    }
}

BookEntry

public class BookEntry {
    private final UUID rentableId = UUID.randomUUID();
    private final UUID personId;
    private final UUID id;
    private final BigDecimal amountDue;
    private final LocalDate paymentDeadline;
    private boolean paid = false;
    private boolean notified = false;

    public BookEntry(UUID personId, UUID id, BigDecimal amountDue, LocalDate paymentDeadline) {
        this.personId = personId;
        this.id = id;
        this.amountDue = amountDue;
        this.paymentDeadline = paymentDeadline;
    }

    public UUID getRentableId() {
        return rentableId;
    }

    public UUID getPersonId() {
        return personId;
    }

    public UUID getId() {
        return id;
    }

    public BigDecimal getAmountDue() {
        return amountDue;
    }

    public LocalDate getPaymentDeadline() {
        return paymentDeadline;
    }

    public boolean isPaid() {
        return paid;
    }

    public boolean isNotified() {
        return notified;
    }

    public void settlePayment() {
        if(!paid) {
            paid = true;
        }
        else {
            throw new IllegalStateException("This is already paid man!");
        }
    }

    public void notifyDebtor() {
        if(!notified) {
            notified = true;
            DataStock dataStock = DataStock.getInstance();
            Person debtor = dataStock.getPeople().stream()
                    .filter(person -> person.getId().equals(personId))
                    .findFirst()
                    .orElseThrow();
            debtor.alert(new TenantAlert(personId, rentableId, dataStock.getCurrentDate(), amountDue));
        }
    }
}

Answer 1

答案似乎很简单——每当 ScheduledExecutorService 中计划的任务抛出异常时，任务就会停止并且不再重复。也不会明显抛出异常。避免这种情况的最简单方法是在运行() 中使用 try-catch 块，即 Runnable 方法。请看看这个post：ScheduledExecutorService handling exceptions

在 ScheduledExecutorService 中使用 Java stream forEach() 冻结

Using Java stream forEach() in ScheduledExecutorService freezes

java

foreach

java-stream

executorservice