共计 3675 个字符,预计需要花费 10 分钟才能阅读完成。
序
本文次要钻研一下 claudb 的 NotificationManager
NotificationManager
claudb-1.7.1/src/main/java/com/github/tonivade/claudb/event/NotificationManager.java
public class NotificationManager implements PatternSubscriptionSupport {
private final DBServerContext server;
private final ExecutorService executor = Executors.newSingleThreadExecutor();
public NotificationManager(DBServerContext server) {this.server = server;}
public void start() {// nothing to do}
public void stop() {executor.shutdown();
}
public void enqueue(Event event) {executor.execute(() -> patternPublish(server, event.getChannel(), event.getValue()));
}
}
- NotificationManager 实现了 PatternSubscriptionSupport 接口,其结构器要求输出 DBServerContext,它提供了 start、stop、enqueue 办法;其 stop 办法执行 executor.shutdown();其 enqueue 办法应用 executor 异步执行 patternPublish(server, event.getChannel(), event.getValue())
patternPublish
claudb-1.7.1/src/main/java/com/github/tonivade/claudb/command/pubsub/PatternSubscriptionSupport.java
public interface PatternSubscriptionSupport extends BaseSubscriptionSupport {
String PSUBSCRIPTION_PREFIX = "psubscription:";
String PMESSAGE = "pmessage";
//......
default int patternPublish(DBServerContext server, String channel, SafeString message) {
int count = 0;
for (Tuple2<String, ImmutableSet<SafeString>> entry : getPatternSubscriptions(server.getAdminDatabase(), channel)) {count += publish(server, entry.get2(), toPatternMessage(entry.get1(), channel, message));
}
return count;
}
default ImmutableSet<Tuple2<String, ImmutableSet<SafeString>>> getPatternSubscriptions(Database admin, String channel) {return getPatternSubscriptions(admin).entries().filter(subscriptionApplyTo(channel));
}
//......
}
- patternPublish 办法遍历 getPatternSubscriptions(server.getAdminDatabase(), channel),挨个执行 publish(server, entry.get2(), toPatternMessage(entry.get1(), channel, message)
BaseSubscriptionSupport
claudb-1.7.1/src/main/java/com/github/tonivade/claudb/command/pubsub/BaseSubscriptionSupport.java
public interface BaseSubscriptionSupport
{default void addSubscription(String suffix, Database admin, String sessionId, SafeString channel) {admin.merge(safeKey(suffix + channel), set(safeString(sessionId)),
(oldValue, newValue) -> set(oldValue.getSet().appendAll(newValue.getSet())));
}
default void removeSubscription(String suffix, Database admin, String sessionId, SafeString channel) {admin.merge(safeKey(suffix + channel), set(safeString(sessionId)),
(oldValue, newValue) -> set(oldValue.getSet().removeAll(newValue.getSet())));
}
default int publish(DBServerContext server, ImmutableSet<SafeString> clients, RedisToken message) {clients.forEach(client -> server.publish(client.toString(), message));
return clients.size();}
}
- BaseSubscriptionSupport 提供了 publish 办法,它遍历 clients,挨个执行 server.publish(client.toString(), message)
publish
claudb-1.7.1/src/main/java/com/github/tonivade/claudb/ClauDB.java
public class ClauDB extends RespServerContext implements DBServerContext {
private static final String STATE = "state";
private static final Logger LOGGER = LoggerFactory.getLogger(ClauDB.class);
private DatabaseCleaner cleaner;
private Option<PersistenceManager> persistence;
private Option<NotificationManager> notifications;
private final DBConfig config;
public ClauDB() {this(DEFAULT_HOST, DEFAULT_PORT);
}
public ClauDB(String host, int port) {this(host, port, DBConfig.builder().build());
}
public ClauDB(String host, int port, DBConfig config) {super(host, port, new DBCommandSuite(), new DBSessionListener());
this.config = config;
}
//......
@Override
public void publish(String sourceKey, RedisToken message) {Session session = getSession(sourceKey);
if (session != null) {session.publish(message);
}
}
//......
}
- ClauDB 的 publish 办法执行的是 session.publish(message)
小结
NotificationManager 实现了 PatternSubscriptionSupport 接口,其结构器要求输出 DBServerContext,它提供了 start、stop、enqueue 办法;其 stop 办法执行 executor.shutdown();其 enqueue 办法应用 executor 异步执行 patternPublish(server, event.getChannel(), event.getValue())
doc
- NotificationManager
正文完