Spring 与 Quartz 完全集成以实现实时通知电子邮件

Question

我目前正在使用 spring boot 开发一个应用程序，让用户创建一个约会。所以基本上约会有一个 startDateTime 和一个 endDateTime 字段 + 一个电子邮件。约会的创建在 MySql 数据库的约会表中添加了一个新行。

我想要做的是在数据库中定义的 startDateTime 之前一小时通过电子邮件通知用户。我寻找了一种解决方案，但找不到。我发现工作（春季批次）可以做到这一点，但工作依赖于频率检查（天、周、月），我正在寻找的是实时通知。欢迎任何有关实现此类任务的解决方案的帮助或指导。

你好

Answer 1

您可以使用诸如quartz 之类的调度库，提供与Spring 框架的轻松集成。

在您的数据库中保存约会后，将在所需时间（例如开始日期前一小时）安排“发送电子邮件”作业。

“发送电子邮件”作业必须实现org.quartz.Job更具体的execute方法，您可以在其中使用您的Autowired SendEmailService实现。

您可以在下面找到一个（几乎）完整的示例，说明如何在代码中实现此类要求。

更新 - 安排作业的代码

首先我们定义一个SchedulingService接口。

public interface SchedulingService {

    startScheduler() throws SchedulerException;

    void standbyScheduler() throws SchedulerException;

    void shutdownScheduler() throws SchedulerException;

    void scheduleJob(JobDetail jobDetail, Trigger trigger) throws SchedulerException;
}

及相关实施。

@Service
public class SchedulingServiceImpl implements SchedulingService {

    @Autowired
    private Scheduler scheduler;

    @Override
    public void startScheduler() throws SchedulerException {
        if (!scheduler.isStarted()) {
            scheduler.start();
        }
    }

    @Override
    public void standbyScheduler() throws SchedulerException {
        if (!scheduler.isInStandbyMode()) {
            scheduler.standby();
        }
    }

    @Override
    public void shutdownScheduler() throws SchedulerException {
        if (!scheduler.isShutdown()) {
            scheduler.shutdown();
        }
    }

    @Override
    public void scheduleJob(JobDetail jobDetail, Trigger trigger) throws SchedulerException {
        scheduler.scheduleJob(jobDetail, trigger);
    }
}

然后在AppointmentServiceImpl我们有一个方法createAppointment()调用scheduleSendEmailJob().

@Service
public class AppointmentServiceImpl implements AppointmentService {

    @Autowired
    private SchedulingService schedulingService;

    public void createAppointment(Appointment appointment) throws SchedulerException {

        // Save appointment to database
        // ...

        // Schedule send email job if appointment has been successfully saved
        scheduleSendEmailJob(appointment);

        return;
    }

    private void scheduleSendEmailJob(Appointment appointment) throws SchedulerException {

        JobDetail jobDetail = JobBuilder.newJob().ofType(SendEmailJob.class)
            .storeDurably()
            .withIdentity(UuidUtils.generateId(), "APPOINTMENT_NOTIFICATIONS")
            .withDescription("Send email notification for appointment")
            .build();


        jobDetail.getJobDataMap().put("appointmentId", appointment.getId());

        Date scheduleDate = appointment.computeDesiredScheduleDate();
        String cronExpression = convertDateToCronExpression(scheduleDate);

        CronTrigger trigger = TriggerBuilder.newTrigger().forJob(jobDetail)
            .withIdentity(UuidUtils.generateId(), "APPOINTMENT_NOTIFICATIONS")
            .withDescription("Trigger description")
            .withSchedule(CronScheduleBuilder.cronSchedule(cronExpression))
            .build();

        schedulingService.scheduleJob(jobDetail, trigger);
    }

    private String convertDateToCronExpression(Date date) {

        Calendar calendar = new GregorianCalendar();

        if (date == null) return null;

        calendar.setTime(date);

        int year = calendar.get(java.util.Calendar.YEAR);
        int month = calendar.get(java.util.Calendar.MONTH) + 1;
        int day = calendar.get(java.util.Calendar.DAY_OF_MONTH);
        int hour = calendar.get(java.util.Calendar.HOUR_OF_DAY);
        int minute = calendar.get(java.util.Calendar.MINUTE);

        return String.format("0 %d %d %d %d ? %d", minute, hour, day, month, year);
    }
}

类SendEmailJob是Job接口的一个实现，负责使用相关服务发送邮件。

更新 - 将参数从调度方法传递到实际作业执行的代码

对于传递参数，正在使用 jobDataMap。例如：

public class SendEmailJob implements Job {

    @Autowired
    private AppointmentService appointmentService;

    @Autowired
    private SendEmailService sendEmailService;

    @Override
    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {

        JobDataMap jobDataMap = jobExecutionContext.getJobDetail().getJobDataMap();

        // Retrieving passed parameters
        Long appointmentId = (Long) jobDataMap.get("appointmentId");

        Appointment appointment = appointmentService.findById(appointmentId);

        // Send email
        sendEmailService.sendEmail(appointment);
    }
}

注意： Appointment 对象也可以从调度方法传递到实际作业执行，你可以只传递：

jobDetail.getJobDataMap().put("appointment", appointment);

并得到：

// Retrieving passed parameters
Appointment appointment = (Appointment) jobDataMap.get("appointment");

更新 - 配置代码

Beanscheduler定义在一个@Configuration负责 Quartz 初始化的类中。

SchedulingConfiguration 类定义为：

@Configuration
public class SchedulingConfiguration {

    @Autowired
    private ApplicationContext applicationContext;

    @Bean
    public Scheduler scheduler() throws SchedulerException, IOException {

        StdSchedulerFactory factory = new StdSchedulerFactory();
        factory.initialize(new ClassPathResource("properties/quartz.properties").getInputStream());

        Scheduler scheduler = factory.getScheduler();
        scheduler.setJobFactory(springBeanJobFactory());

        return scheduler;
    }

    @Bean
    public SpringBeanJobFactory springBeanJobFactory() {
        AutowiringSpringBeanJobFactory jobFactory = new AutowiringSpringBeanJobFactory();
        jobFactory.setApplicationContext(applicationContext);
        return jobFactory;
    }

}

我们的quartz.properties文件位于resources/properties文件夹中。请注意，作业持久性数据库是一个 Oracle 实例。

# Configure Main Scheduler Properties
org.quartz.scheduler.instanceName = AppScheduler
org.quartz.scheduler.instanceId = AUTO

# Configure ThreadPool
org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool
org.quartz.threadPool.threadCount = 10
org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread = true

# Configure JobStore
org.quartz.jobStore.misfireThreshold = 60000
org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.jobStore.driverDelegateClass = 
org.quartz.impl.jdbcjobstore.oracle.OracleDelegate
org.quartz.jobStore.tablePrefix = APP.QRTZ_
org.quartz.jobStore.useProperties = false
org.quartz.jobStore.dataSource = appDs
org.quartz.jobStore.isClustered = true
org.quartz.jobStore.clusterCheckinInterval = 20000

# Configure Datasources
org.quartz.dataSource.appDs.driver = oracle.jdbc.driver.OracleDriver
org.quartz.dataSource.appDs.URL = jdbc:oracle:thin:@dbsrv:1521:appdb
org.quartz.dataSource.appDs.user = db_user
org.quartz.dataSource.appDs.password = db_pwd
org.quartz.dataSource.appDs.maxConnections = 5
org.quartz.dataSource.appDs.validationQuery = select 0 from dual

最后一步是在应用程序上下文初始化中调用调度程序方法如下（请注意在 中添加的方法SchedulingService）：

public class SchedulingContextListener implements ServletContextListener {

    private static final Logger logger = LogManager.getLogger(SchedulingContextListener.class);

    private SchedulingService schedulingService(ServletContextEvent sce) {
        WebApplicationContext springContext = WebApplicationContextUtils.getWebApplicationContext(sce.getServletContext());
        return springContext.getBean(SchedulingService.class);
    }

    @Override
    public void contextInitialized(ServletContextEvent sce) {
        try {
            this.schedulingService(sce).startScheduler();
        } catch (SchedulerException e) {
            logger.error("Error while Scheduler is being started", e);
        }
    }

    @Override
    public void contextDestroyed(ServletContextEvent sce) {
        try {
            this.schedulingService(sce).shutdownScheduler();
        } catch (SchedulerException e) {
            logger.error("Error while Scheduler is being shutdown", e);
        }
    }
}

注意： SchedulingContextListener应该servletContext在应用程序初始化中注册，具体取决于 Spring 配置是如何定义的，使用 Spring Boot 或传统的 Spring MVC 配置。

希望有帮助。