本文目录导读：
在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
}
}
最佳实践建议
- 异步处理：回调操作建议使用线程池异步执行，避免阻塞监听线程
- 异常处理：回调中必须处理异常，防止影响其他监听器
- 幂等性设计：回调操作应该支持重复执行
- 超时控制：设置回调执行超时时间
- 顺序保证：如果业务需要保证顺序，使用同步回调队列
// 异步回调执行器示例
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适合微服务配置管理。
