SpringBoot文章定时发布的技术方案详解(三种延迟任务方案)

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class ArticleTask implements Delayed {
    private final long executeTime; // 执行时间戳
    private final Article article; // 文章内容
    public ArticleTask(Article article, long delay) {
        this.article = article;
        this.executeTime = System.currentTimeMillis() + delay;
    }
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(executeTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    @Override
    public int compareTo(Delayed o) {
        return Long.compare(this.executeTime, ((ArticleTask) o).executeTime);
    }
    // Getter方法
    public Article getArticle() {
        return article;
    }
}

public class Article {
    private Long id;
    private String title;
    private String content;
    private Integer status; // 状态：0-草稿，1-已发布
    private Date publishTime; // 预定发布时间
    // 省略getter/setter
}

@Component
public class DelayQueueService {
    private final DelayQueue<ArticleTask> queue = new DelayQueue<>();
    @PostConstruct
    public void init() {
        // 启动消费线程
        new Thread(this::consume).start();
    }
    /**
     * 添加定时发布任务
     * @param article 文章
     * @param delay 延迟时间（毫秒）
     */
    public void addTask(Article article, long delay) {
        queue.put(new ArticleTask(article, delay));
    }
    /**
     * 消费延迟队列中的任务
     */
    private void consume() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                ArticleTask task = queue.take(); // 阻塞直到有任务到期
                publishArticle(task.getArticle());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    /**
     * 发布文章
     */
    private void publishArticle(Article article) {
        // 更新文章状态为已发布
        article.setStatus(1);
        article.setPublishTime(new Date());
        System.out.println("发布时间：" + new Date() + "，发布文章: " + article.getTitle());
        // 这里可以加入实际的数据库更新逻辑
        // articleService.updateById(article);
    }
}

@RestController
@RequestMapping("/article")
public class ArticleController {
    @Autowired
    private DelayQueueService delayQueueService;
    @PostMapping("/schedule")
    public String schedulePublish(@RequestBody Article article, 
                                 @RequestParam long delayMillis) {
        delayQueueService.addTask(article, delayMillis);
        return "文章已安排定时发布";
    }
}

@Configuration
public class RabbitMQConfig {
    // 定义死信交换机和队列
    @Bean
    public DirectExchange deadLetterExchange() {
        return new DirectExchange("dead.letter.exchange");
    }
    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable("dead.letter.queue").build();
    }
    @Bean
    public Binding deadLetterBinding() {
        return BindingBuilder.bind(deadLetterQueue())
                .to(deadLetterExchange())
                .with("dead.letter");
    }
    // 定义实际业务队列，并绑定死信交换机
    @Bean
    public Queue articleQueue() {
        return QueueBuilder.durable("article.delay.queue")
                .withArgument("x-dead-letter-exchange", "dead.letter.exchange")
                .withArgument("x-dead-letter-routing-key", "dead.letter")
                .build();
    }
}

@Service
public class ArticleMQService {
    @Autowired
    private RabbitTemplate rabbitTemplate;
    /**
     * 安排文章定时发布
     * @param article 文章
     * @param delayMillis 延迟时间（毫秒）
     */
    public void scheduleArticle(Article article, long delayMillis) {
        rabbitTemplate.convertAndSend("", "article.delay.queue", article, message -> {
            // 设置消息的过期时间
            message.getMessageProperties().setExpiration(String.valueOf(delayMillis));
            return message;
        });
        System.out.println("消息已发送，将在 " + delayMillis + " 毫秒后过期");
    }
}

@Component
public class ArticleDelayListener {
    @RabbitListener(queues = "dead.letter.queue")
    public void processDelayedArticle(Article article) {
        // 此时文章已到预定发布时间，执行发布逻辑
        article.setStatus(1);
        article.setPublishTime(new Date());
        System.out.println("发布时间：" + new Date() + "，发布文章: " + article.getTitle());
        // 实际业务中，这里应该更新数据库
        // articleService.updateById(article);
    }
}

@RestController
@RequestMapping("/article")
public class ArticleController {
    @Autowired
    private ArticleMQService articleMQService;
    @PostMapping("/schedule")
    public String schedulePublish(@RequestBody Article article, 
                                 @RequestParam long delayMillis) {
        articleMQService.scheduleArticle(article, delayMillis);
        return "文章已安排定时发布";
    }
}

@Service
public class RedisDelayService {
    @Autowired
    private StringRedisTemplate redisTemplate;
    private static final String DELAY_KEY = "article:delay:tasks";
    // 使用Jackson进行序列化
    private final ObjectMapper objectMapper = new ObjectMapper();
    /**
     * 添加延迟任务
     */
    public void addArticleTask(Article article, long delaySeconds) {
        long executeTime = System.currentTimeMillis() + (delaySeconds * 1000);
        try {
            String articleJson = objectMapper.writeValueAsString(article);
            redisTemplate.opsForZSet().add(DELAY_KEY, articleJson, executeTime);
        } catch (JsonProcessingException e) {
            throw new RuntimeException("文章序列化失败", e);
        }
    }
    /**
     * 获取到期的任务
     */
    public Set<String> getExpiredTasks(long maxScore) {
        return redisTemplate.opsForZSet().rangeByScore(DELAY_KEY, 0, maxScore);
    }
    /**
     * 移除已处理的任务
     */
    public void removeTask(String task) {
        redisTemplate.opsForZSet().remove(DELAY_KEY, task);
    }
}

@Component
public class RedisTaskScanner {
    @Autowired
    private RedisDelayService redisDelayService;
    private static final String LOCK_KEY = "article:delay:lock";
    private static final long LOCK_EXPIRE = 30; // 锁过期时间30秒
    /**
     * 每秒扫描一次到期任务
     */
    @Scheduled(fixedRate = 1000)
    public void scanExpiredTasks() {
        // 使用Redis分布式锁，防止集群环境下重复消费
        boolean locked = tryGetDistributedLock();
        if (!locked) {
            return;
        }
        try {
            long maxScore = System.currentTimeMillis();
            Set<String> expiredTasks = redisDelayService.getExpiredTasks(maxScore);
            if (expiredTasks != null && !expiredTasks.isEmpty()) {
                for (String taskJson : expiredTasks) {
                    processTask(taskJson);
                    // 从集合中移除已处理的任务
                    redisDelayService.removeTask(taskJson);
                }
            }
        } finally {
            // 释放锁
            releaseDistributedLock();
        }
    }
    private void processTask(String taskJson) {
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            Article article = objectMapper.readValue(taskJson, Article.class);
            // 执行发布逻辑
            article.setStatus(1);
            article.setPublishTime(new Date());
            System.out.println("发布时间：" + new Date() + "，发布文章: " + article.getTitle());
            // 实际业务中更新数据库
            // articleService.updateById(article);
        } catch (Exception e) {
            System.err.println("处理任务失败: " + e.getMessage());
            // 可以加入重试机制或死信队列
        }
    }
    private boolean tryGetDistributedLock() {
        // 简化的分布式锁实现，生产环境建议使用Redisson等成熟方案
        // 这里只是示意，实际实现需要考虑原子性等问题
        return true;
    }
    private void releaseDistributedLock() {
        // 释放锁的实现
    }
}

@RestController
@RequestMapping("/article")
public class ArticleController {
    @Autowired
    private RedisDelayService redisDelayService;
    @PostMapping("/schedule")
    public String schedulePublish(@RequestBody Article article, 
                                 @RequestParam long delaySeconds) {
        redisDelayService.addArticleTask(article, delaySeconds);
        return "文章已安排定时发布";
    }
}

@SpringBootApplication
@EnableScheduling // 开启定时任务支持
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

@Component
public class DatabaseScheduler {
    @Autowired
    private ArticleService articleService;
    // 每分钟执行一次
    @Scheduled(fixedRate = 60000)
    public void publishScheduledArticles() {
        List<Article> articles = articleService.findScheduledArticles(new Date());
        for (Article article : articles) {
            article.setStatus(1);
            articleService.update(article);
            System.out.println("发布文章: " + article.getTitle());
        }
    }
}