如何将后台线程(T1)中抛出的异常通知给调度程序线程(T2)

Question

JobRunner.Class

    @Scheduled
    public void pipelineRunner() {
        List<Job> jobs = getJobs(); //fetches tasks from database
        for(Job job: jobs){
            try{
                heartBeatSender.startHeartBeat(job); // updates last_update_time column for the current job in database on background thread
                runJob(job); // contains logic of processing a job, how to abort current job  it is processing if heartbeat sender throws a runtime exception or how to notify or make scheduler thread throw exception so that it can move to next job in the list?
            }catch(Exception e){}
            finally{
                heartBeatSender.stopHeartBeat(); //stops updating last_update_time col.
            }
        }
    }

heartbeatsender.class

  
    public void startHeartBeat(Job job) {
        Runnable heartBeat = () -> {
            if (job != null) {
                job.setLastUpdateTime(System.currentTimeMillis());
                jobRepository.save(job);//runtime exception can be thrown here.
            }
        }
        executor = Executors.newSingleThreadScheduledExecutor();
        executor.scheduleAtFixedRate(heartBeat, 0L, heartBeatInterval, TimeUnit.MILLISECONDS);  //heartBeatInterval every 5mins
    }

Answer 1

您发布的代码有点不一致，因为heartbeatsender似乎可以向多个作业发送心跳，但客户端代码总是只触发一个。但是，如果您有另一个类也使用这个heartbeatsender，您将已经安排了两个heartbeat运行表，那么它如何决定将取消哪个stopheartbeat()？更不用说，我一开始就无法知道它是如何取消它的，因为它不跟踪startheartbeat()调用之外的计划任务。

所以您可能想要做的是让heartbeatsender按作业跟踪它的心跳，这样它就可以单独取消它们。

class HeartbeatSender {
  private final Map<Job, ScheduledFuture<?>> heartbeats = new HashMap<>();
  // you can create a new executor in start, but I don't see why
  private final ScheduledExecutorService heartbeatExecutor = newSingleThreadScheduledExecutor();

  public void startHeartbeat(Job job) {
    Runnable heartBeat= ...;
    ScheduledFuture<?> schedule = executor.scheduleAtFixedRate(heartBeat, 0L, heartBeatInterval, TimeUnit.MILLISECONDS);
    heartbeats.put(job, schedule); // remember so you can cancel later
  }

  public void stopHeartbeat(Job job) {
    ScheduledFuture<?> schedule = heartbeats.remove(job);
    if (schedule != null) {
      schedule.cancel(false);
    }
  }
}

现在，您可以按作业停止心跳，所以您只需要在客户端代码中这样做。

for(Job job: jobs){
  try {
    heartBeatSender.startHeartBeat(job);
    runJob(job);
  } catch(Exception e) {
  } finally{
    heartBeatSender.stopHeartBeat(job); // stops heartbeat for this job
  }
}