如何实现Spring异步通知发送应用的定时检查与发送功能？

数据老陈

2026-7-8

基于Spring的异步通知系统实现方案

一、第一个服务：接收前端数据并写入数据库

这个服务的核心是接收前端提交的通知参数，持久化到数据库中，需要设计合理的数据库实体存储关键信息。

1. 数据库实体设计

创建Notification实体类，包含以下核心字段：

id：主键ID
content：通知内容
triggerTime：触发发送的时间
status：通知状态（比如PENDING待发送、SENT已发送、FAILED发送失败）
createTime：创建时间

2. 代码实现示例

Controller层（接收前端请求）

@RestController
@RequestMapping("/api/notifications")
public class NotificationController {
    private final NotificationService notificationService;

    public NotificationController(NotificationService notificationService) {
        this.notificationService = notificationService;
    }

    @PostMapping
    public ResponseEntity<Void> submitNotification(@RequestBody NotificationRequest request) {
        notificationService.saveNotification(request);
        return ResponseEntity.status(HttpStatus.CREATED).build();
    }
}

Service层（业务逻辑处理）

@Service
public class NotificationService {
    private final NotificationRepository notificationRepository;

    public NotificationService(NotificationRepository notificationRepository) {
        this.notificationRepository = notificationRepository;
    }

    public void saveNotification(NotificationRequest request) {
        Notification notification = new Notification();
        notification.setContent(request.getContent());
        notification.setTriggerTime(request.getTriggerTime());
        notification.setStatus(NotificationStatus.PENDING);
        notification.setCreateTime(LocalDateTime.now());
        notificationRepository.save(notification);
    }
}

Repository层（数据库操作）

public interface NotificationRepository extends JpaRepository<Notification, Long> {
    @Query("SELECT n FROM Notification n WHERE n.triggerTime <= :currentTime AND n.status = :status FOR UPDATE")
    List<Notification> findPendingNotifications(@Param("currentTime") LocalDateTime currentTime, 
                                                @Param("status") NotificationStatus status);
}

二、第二个服务：定时检查并发送通知

提供三种主流实现方案，可根据业务规模和需求选择：

方案1：Spring原生@Scheduled定时任务

适合中小规模场景，实现简单，无需额外依赖。

1. 开启定时任务支持

在启动类上添加@EnableScheduling注解：

@SpringBootApplication
@EnableScheduling
public class NotificationSchedulerApplication {
    public static void main(String[] args) {
        SpringApplication.run(NotificationSchedulerApplication.class, args);
    }
}

2. 编写定时任务逻辑

@Service
public class NotificationScheduler {
    private static final Logger log = LoggerFactory.getLogger(NotificationScheduler.class);
    private final NotificationRepository notificationRepository;
    private final NotificationSender notificationSender;

    public NotificationScheduler(NotificationRepository notificationRepository, 
                                 NotificationSender notificationSender) {
        this.notificationRepository = notificationRepository;
        this.notificationSender = notificationSender;
    }

    // 每分钟执行一次，可根据需求调整cron表达式
    @Scheduled(cron = "0 */1 * * * ?")
    public void checkAndSendNotifications() {
        LocalDateTime now = LocalDateTime.now();
        // 加行锁避免多实例并发时重复处理
        List<Notification> pendingNotifications = notificationRepository.findPendingNotifications(now, NotificationStatus.PENDING);
        
        for (Notification notification : pendingNotifications) {
            try {
                // 调用发送逻辑，比如调用其他服务的API
                notificationSender.send(notification);
                notification.setStatus(NotificationStatus.SENT);
            } catch (Exception e) {
                // 发送失败标记状态，后续可重试
                notification.setStatus(NotificationStatus.FAILED);
                log.error("Failed to send notification: {}", notification.getId(), e);
            }
            notificationRepository.save(notification);
        }
    }
}

注意事项

多实例部署时需添加分布式锁（如基于Redis的Redisson），避免多个实例重复执行任务。
cron表达式的间隔决定了最大延迟时间，比如1分钟间隔，通知最多会延迟1分钟发送。

方案2：Quartz定时任务框架

适合复杂定时需求（如动态修改触发时间、大量定时任务），支持分布式部署。

1. 引入依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>

2. 核心实现逻辑

第一个服务写入数据库时，同时创建Quartz任务和触发器：

@Service
public class NotificationService {
    private final NotificationRepository notificationRepository;
    private final Scheduler scheduler;

    public NotificationService(NotificationRepository notificationRepository, Scheduler scheduler) {
        this.notificationRepository = notificationRepository;
        this.scheduler = scheduler;
    }

    public void saveNotification(NotificationRequest request) throws SchedulerException {
        // 保存数据库记录
        Notification notification = new Notification();
        notification.setContent(request.getContent());
        notification.setTriggerTime(request.getTriggerTime());
        notification.setStatus(NotificationStatus.PENDING);
        notification.setCreateTime(LocalDateTime.now());
        notificationRepository.save(notification);

        // 创建Quartz任务
        JobDetail jobDetail = JobBuilder.newJob(NotificationSendJob.class)
                .withIdentity("notification-job-" + notification.getId())
                .usingJobData("notificationId", notification.getId())
                .build();

        // 设置触发时间
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity("notification-trigger-" + notification.getId())
                .startAt(Date.from(request.getTriggerTime().atZone(ZoneId.systemDefault()).toInstant()))
                .build();

        scheduler.scheduleJob(jobDetail, trigger);
    }
}

3. 编写Job执行逻辑

public class NotificationSendJob implements Job {
    private static final Logger log = LoggerFactory.getLogger(NotificationSendJob.class);
    private final NotificationRepository notificationRepository;
    private final NotificationSender notificationSender;

    public NotificationSendJob(NotificationRepository notificationRepository, 
                               NotificationSender notificationSender) {
        this.notificationRepository = notificationRepository;
        this.notificationSender = notificationSender;
    }

    @Override
    public void execute(JobExecutionContext context) throws JobExecutionException {
        Long notificationId = context.getJobDetail().getJobDataMap().getLong("notificationId");
        Notification notification = notificationRepository.findById(notificationId)
                .orElseThrow(() -> new JobExecutionException("Notification not found: " + notificationId));

        if (notification.getStatus() == NotificationStatus.PENDING) {
            try {
                notificationSender.send(notification);
                notification.setStatus(NotificationStatus.SENT);
            } catch (Exception e) {
                notification.setStatus(NotificationStatus.FAILED);
                log.error("Failed to send notification in Quartz job: {}", notificationId, e);
            }
            notificationRepository.save(notification);
        }
    }
}

方案3：延迟消息队列（如RabbitMQ/RocketMQ）

适合高并发、低延迟场景，无需轮询数据库，资源消耗更低。

1. 核心逻辑

第一个服务写入数据库后，发送一条延迟消息到MQ，延迟时间为triggerTime - 当前时间；第二个服务作为消费者，收到消息后执行发送逻辑。

2. 示例（RabbitMQ延迟队列）

配置延迟交换机和队列
发送延迟消息：

@Service
public class NotificationService {
    private final NotificationRepository notificationRepository;
    private final RabbitTemplate rabbitTemplate;

    public NotificationService(NotificationRepository notificationRepository, RabbitTemplate rabbitTemplate) {
        this.notificationRepository = notificationRepository;
        this.rabbitTemplate = rabbitTemplate;
    }

    public void saveNotification(NotificationRequest request) {
        Notification notification = new Notification();
        notification.setContent(request.getContent());
        notification.setTriggerTime(request.getTriggerTime());
        notification.setStatus(NotificationStatus.PENDING);
        notification.setCreateTime(LocalDateTime.now());
        notificationRepository.save(notification);

        // 计算延迟时间（毫秒）
        long delay = Duration.between(LocalDateTime.now(), request.getTriggerTime()).toMillis();
        // 发送延迟消息
        rabbitTemplate.convertAndSend("delay-exchange", "delay-routing-key", notification.getId(), message -> {
            message.getMessageProperties().setDelay((int) delay);
            return message;
        });
    }
}

消费者接收消息并发送：

@Component
public class NotificationConsumer {
    private static final Logger log = LoggerFactory.getLogger(NotificationConsumer.class);
    private final NotificationRepository notificationRepository;
    private final NotificationSender notificationSender;

    public NotificationConsumer(NotificationRepository notificationRepository, 
                                NotificationSender notificationSender) {
        this.notificationRepository = notificationRepository;
        this.notificationSender = notificationSender;
    }

    @RabbitListener(queues = "delay-queue")
    public void handleNotification(Long notificationId) {
        Notification notification = notificationRepository.findById(notificationId)
                .orElseThrow(() -> new RuntimeException("Notification not found: " + notificationId));

        if (notification.getStatus() == NotificationStatus.PENDING) {
            try {
                notificationSender.send(notification);
                notification.setStatus(NotificationStatus.SENT);
            } catch (Exception e) {
                notification.setStatus(NotificationStatus.FAILED);
                log.error("Failed to send notification from MQ: {}", notificationId, e);
            }
            notificationRepository.save(notification);
        }
    }
}

三、通用注意事项

幂等性保障：通过status字段判断是否已处理，避免重复发送；也可给每条通知生成唯一标识，发送时校验。
重试机制：发送失败的通知，可通过定时任务重新触发，或存入死信队列人工干预。
分布式场景：如果是微服务架构，需考虑分布式锁、分布式定时任务（如XXL-JOB）来避免重复执行。

内容的提问来源于stack exchange，提问作者Kadri