关于redis:聊聊claudb的NotificationManager

序

本文次要钻研一下 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