import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public class ArticleTask implements Delayed {
    private final long executeTime; // 執(zhí)行時間戳
    private final Article article; // 文章內(nèi)容
    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; // 狀態(tài)：0-草稿，1-已發(fā)布
    private Date publishTime; // 預(yù)定發(fā)布時間
    // 省略getter/setter
}

@Component
public class DelayQueueService {
    private final DelayQueue<ArticleTask> queue = new DelayQueue<>();
    @PostConstruct
    public void init() {
        // 啟動消費線程
        new Thread(this::consume).start();
    }
    /**
     * 添加定時發(fā)布任務(wù)
     * @param article 文章
     * @param delay 延遲時間（毫秒）
     */
    public void addTask(Article article, long delay) {
        queue.put(new ArticleTask(article, delay));
    }
    /**
     * 消費延遲隊列中的任務(wù)
     */
    private void consume() {
        while (!Thread.currentThread().isInterrupted()) {
            try {
                ArticleTask task = queue.take(); // 阻塞直到有任務(wù)到期
                publishArticle(task.getArticle());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
            }
        }
    }
    /**
     * 發(fā)布文章
     */
    private void publishArticle(Article article) {
        // 更新文章狀態(tài)為已發(fā)布
        article.setStatus(1);
        article.setPublishTime(new Date());
        System.out.println("發(fā)布時間：" + new Date() + "，發(fā)布文章: " + article.getTitle());
        // 這里可以加入實際的數(shù)據(jù)庫更新邏輯
        // 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 "文章已安排定時發(fā)布";
    }
}

@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");
    }
    // 定義實際業(yè)務(wù)隊列，并綁定死信交換機
    @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;
    /**
     * 安排文章定時發(fā)布
     * @param article 文章
     * @param delayMillis 延遲時間（毫秒）
     */
    public void scheduleArticle(Article article, long delayMillis) {
        rabbitTemplate.convertAndSend("", "article.delay.queue", article, message -> {
            // 設(shè)置消息的過期時間
            message.getMessageProperties().setExpiration(String.valueOf(delayMillis));
            return message;
        });
        System.out.println("消息已發(fā)送，將在 " + delayMillis + " 毫秒后過期");
    }
}

@Component
public class ArticleDelayListener {
    @RabbitListener(queues = "dead.letter.queue")
    public void processDelayedArticle(Article article) {
        // 此時文章已到預(yù)定發(fā)布時間，執(zhí)行發(fā)布邏輯
        article.setStatus(1);
        article.setPublishTime(new Date());
        System.out.println("發(fā)布時間：" + new Date() + "，發(fā)布文章: " + article.getTitle());
        // 實際業(yè)務(wù)中，這里應(yīng)該更新數(shù)據(jù)庫
        // 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 "文章已安排定時發(fā)布";
    }
}

@Service
public class RedisDelayService {
    @Autowired
    private StringRedisTemplate redisTemplate;
    private static final String DELAY_KEY = "article:delay:tasks";
    // 使用Jackson進行序列化
    private final ObjectMapper objectMapper = new ObjectMapper();
    /**
     * 添加延遲任務(wù)
     */
    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);
        }
    }
    /**
     * 獲取到期的任務(wù)
     */
    public Set<String> getExpiredTasks(long maxScore) {
        return redisTemplate.opsForZSet().rangeByScore(DELAY_KEY, 0, maxScore);
    }
    /**
     * 移除已處理的任務(wù)
     */
    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秒
    /**
     * 每秒掃描一次到期任務(wù)
     */
    @Scheduled(fixedRate = 1000)
    public void scanExpiredTasks() {
        // 使用Redis分布式鎖，防止集群環(huán)境下重復(fù)消費
        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);
                    // 從集合中移除已處理的任務(wù)
                    redisDelayService.removeTask(taskJson);
                }
            }
        } finally {
            // 釋放鎖
            releaseDistributedLock();
        }
    }
    private void processTask(String taskJson) {
        try {
            ObjectMapper objectMapper = new ObjectMapper();
            Article article = objectMapper.readValue(taskJson, Article.class);
            // 執(zhí)行發(fā)布邏輯
            article.setStatus(1);
            article.setPublishTime(new Date());
            System.out.println("發(fā)布時間：" + new Date() + "，發(fā)布文章: " + article.getTitle());
            // 實際業(yè)務(wù)中更新數(shù)據(jù)庫
            // articleService.updateById(article);
        } catch (Exception e) {
            System.err.println("處理任務(wù)失敗: " + e.getMessage());
            // 可以加入重試機制或死信隊列
        }
    }
    private boolean tryGetDistributedLock() {
        // 簡化的分布式鎖實現(xiàn)，生產(chǎn)環(huán)境建議使用Redisson等成熟方案
        // 這里只是示意，實際實現(xiàn)需要考慮原子性等問題
        return true;
    }
    private void releaseDistributedLock() {
        // 釋放鎖的實現(xiàn)
    }
}

@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 "文章已安排定時發(fā)布";
    }
}

@SpringBootApplication
@EnableScheduling // 開啟定時任務(wù)支持
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

@Component
public class DatabaseScheduler {
    @Autowired
    private ArticleService articleService;
    // 每分鐘執(zhí)行一次
    @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("發(fā)布文章: " + article.getTitle());
        }
    }
}