协调多个生产者之间的定期任务

在多个生产者之间协调定期任务，可以使用分布式任务调度工具如Apache ZooKeeper和Spring Scheduler。

一个示例代码：

使用Apache ZooKeeper：

public class DistributedJobScheduler {

    private static final String ZK_ADDRESS = "localhost:2181";
    private static final String JOB_PATH = "/jobs";
    private static final byte[] DATA = "job data".getBytes();
    private static final int INITIAL_DELAY = 0;
    private static final int PERIOD = 10000;

    public static void main(String[] args) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory.newClient(ZK_ADDRESS, new RetryNTimes(10, 500));
        client.start();

        ExecutorService executor = Executors.newFixedThreadPool(10);
        for (int i = 0; i < 10; i++) {
            int producerId = i;
            executor.submit(() -> {
                try {
                    String jobPath = JOB_PATH + "/producer" + producerId;
                    addJob(client, jobPath, DATA, INITIAL_DELAY, PERIOD);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        executor.shutdown();
        executor.awaitTermination(10, TimeUnit.MINUTES);

        client.close();
    }

    private static void addJob(CuratorFramework client, String path, byte[] data, int initialDelay, int period) throws Exception {
        if (client.checkExists().forPath(path) == null) {
            client.create().withMode(CreateMode.PERSISTENT).forPath(path, data);
        }

        ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
        scheduler.scheduleAtFixedRate(() -> {
            System.out.println("Running job at " + Instant.now() + " for " + path);
        }, initialDelay, period, TimeUnit.MILLISECONDS);

        ScheduledFuture<?> future = scheduler.scheduleAtFixedRate(() -> {
            try {
                client.setData().forPath(path, data);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, initialDelay, period, TimeUnit.MILLISECONDS);

        while (!future.isDone()) {
            Thread.sleep(1000);
        }

        scheduler.shutdown();
    }

}

使用Spring Scheduler：

@Configuration
@EnableScheduling
public class JobConfiguration {

    private static final byte[] DATA = "job data".getBytes();
    private static final int INITIAL_DELAY = 0;
    private static final int PERIOD = 10000;

    @Autowired
    private CuratorFramework client;

    @Bean
    public List<Job> jobs() {
        List<Job> jobs = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            int producerId