Java分布式数据监听器怎么回调

本文目录导读：

1. 基于Zookeeper的回调实现
2. 基于Redis的发布订阅回调
3. 基于消息队列的回调（Kafka示例）
4. 自定义分布式监听器框架
5. Spring Cloud Config的回调
6. 最佳实践建议

在Java分布式系统中,数据监听器的回调机制通常涉及以下几个关键部分：

基于Zookeeper的回调实现

Watcher回调机制

public class ZKDataListener implements Watcher {
    private ZooKeeper zooKeeper;
    private String watchPath;
    public ZKDataListener(String connectString, String watchPath) throws Exception {
        this.watchPath = watchPath;
        this.zooKeeper = new ZooKeeper(connectString, 3000, this);
        // 监听数据变化
        zooKeeper.getData(watchPath, this, null);
    }
    @Override
    public void process(WatchedEvent event) {
        if (event.getType() == Event.EventType.NodeDataChanged) {
            // 数据变更回调
            handleDataChange(event.getPath());
            // 重新注册监听（Zookeeper是一次性监听）
            try {
                zooKeeper.getData(watchPath, this, null);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
    private void handleDataChange(String path) {
        try {
            byte[] data = zooKeeper.getData(path, false, null);
            String newValue = new String(data);
            System.out.println("数据变更: " + path + " -> " + newValue);
            // 执行你的业务逻辑
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

基于Redis的发布订阅回调

Redis Pub/Sub模式

public class RedisDataListener implements MessageListener {
    private JedisPool jedisPool;
    public RedisDataListener() {
        this.jedisPool = new JedisPool("localhost", 6379);
        startListening();
    }
    public void startListening() {
        new Thread(() -> {
            try (Jedis jedis = jedisPool.getResource()) {
                // 订阅频道
                jedis.subscribe(new JedisPubSub() {
                    @Override
                    public void onMessage(String channel, String message) {
                        // 回调方法
                        handleDataChange(channel, message);
                    }
                    @Override
                    public void onSubscribe(String channel, int subscribedChannels) {
                        System.out.println("订阅成功: " + channel);
                    }
                }, "data_channel");
            }
        }).start();
    }
    private void handleDataChange(String channel, String message) {
        System.out.println("数据变更通知: " + message);
        // 处理数据变更业务逻辑
    }
}

基于消息队列的回调（Kafka示例）

Kafka消费者回调

public class KafkaDataListener {
    private KafkaConsumer<String, String> consumer;
    public KafkaDataListener() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "localhost:9092");
        props.put("group.id", "data-listener-group");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        this.consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList("data-topic"));
    }
    public void startListening() {
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            for (ConsumerRecord<String, String> record : records) {
                // 回调处理
                handleDataChange(record.key(), record.value());
            }
        }
    }
    private void handleDataChange(String key, String value) {
        System.out.printf("数据变更 - key: %s, value: %s%n", key, value);
        // 业务逻辑处理
    }
}

自定义分布式监听器框架

public class DistributedDataListener {
    private final ExecutorService callbackExecutor;
    private final List<DataChangeCallback> callbacks;
    public DistributedDataListener() {
        this.callbackExecutor = Executors.newCachedThreadPool();
        this.callbacks = new CopyOnWriteArrayList<>();
    }
    // 注册回调
    public void registerCallback(DataChangeCallback callback) {
        callbacks.add(callback);
    }
    // 触发回调
    public void notifyDataChange(String dataKey, String oldValue, String newValue) {
        callbackExecutor.submit(() -> {
            for (DataChangeCallback callback : callbacks) {
                try {
                    callback.onDataChange(dataKey, oldValue, newValue);
                } catch (Exception e) {
                    // 异常处理，避免影响其他回调
                    System.err.println("回调执行失败: " + e.getMessage());
                }
            }
        });
    }
    @FunctionalInterface
    public interface DataChangeCallback {
        void onDataChange(String key, String oldValue, String newValue);
    }
}

Spring Cloud Config的回调

@Component
public class ConfigChangeListener {
    @Autowired
    private ConfigurableApplicationContext applicationContext;
    @EventListener
    public void onConfigChange(EnvironmentChangeEvent event) {
        Set<String> changedKeys = event.getKeys();
        changedKeys.forEach(key -> {
            String newValue = applicationContext.getEnvironment().getProperty(key);
            System.out.println("配置变更: " + key + " = " + newValue);
            // 执行配置变更后的业务逻辑
        });
    }
    // 或者使用特定注解
    @ConfigurationProperties("app.config")
    @RefreshScope
    public class ConfigProperties {
        private String dataKey;
        // getter/setter
    }
}

最佳实践建议

1. 异步处理：回调操作建议使用线程池异步执行，避免阻塞监听线程
2. 异常处理：回调中必须处理异常，防止影响其他监听器
3. 幂等性设计：回调操作应该支持重复执行
4. 超时控制：设置回调执行超时时间
5. 顺序保证：如果业务需要保证顺序，使用同步回调队列

// 异步回调执行器示例
public class AsyncCallbackExecutor {
    private final ThreadPoolExecutor executor = new ThreadPoolExecutor(
        5, 20, 60, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(1000),
        new ThreadPoolExecutor.CallerRunsPolicy()
    );
    public void executeCallback(Runnable callback) {
        CompletableFuture.runAsync(callback, executor)
            .orTimeout(5, TimeUnit.SECONDS)
            .exceptionally(throwable -> {
                System.err.println("回调执行超时或异常: " + throwable.getMessage());
                return null;
            });
    }
}

选择哪种回调机制取决于你的具体需求：Zookeeper适合配置管理，Redis适合实时通知，Kafka适合大数据处理，Spring Cloud Config适合微服务配置管理。