class NotificationDispatcher implements CommandLineRunner {
    private final EventBus eventBus;

    private final ReactiveCacheContainer<String, NotifyChannel> channels = ReactiveCacheContainer.create();

    private final Map<String, NotifyChannelProvider> providers = new HashMap<>();

    private final ReactiveRepository<NotifySubscriberChannelEntity, String> channelRepository;

    public NotificationDispatcher(EventBus eventBus,
                                  ObjectProvider<NotifyChannelProvider> providers,
                                  ReactiveRepository<NotifySubscriberChannelEntity, String> channelRepository) {
        this.eventBus = eventBus;
        this.channelRepository = channelRepository;
        //默认支持站内信
        this.channels.put(InsideMailChannelProvider.provider, new InsideMailChannelProvider(eventBus));

        for (NotifyChannelProvider provider : providers) {
            this.providers.put(provider.getId(), provider);
        }
    }

    @EventListener
    public void handleNotifications(Notification notification) {

        List<String> channelIdList = notification.getNotifyChannels();
        //默认站内信
        if (channelIdList == null) {
            channelIdList = Collections.singletonList(InsideMailChannelProvider.provider);
        }
        //发送通知
        for (String notifyChannel : channelIdList) {
            NotifyChannel dispatcher = channels.getNow(notifyChannel);
            if (dispatcher != null) {
                dispatcher
                    .sendNotify(notification)
                    .subscribe();
            }
        }

    }

    @EventListener
    public void handleEvent(EntityCreatedEvent<NotifySubscriberChannelEntity> event) {

        event.async(
            register(event.getEntity())
        );
    }

    @EventListener
    public void handleEvent(EntitySavedEvent<NotifySubscriberChannelEntity> event) {

        event.async(
            register(event.getEntity())
        );
    }

    @EventListener
    public void handleEvent(EntityModifyEvent<NotifySubscriberChannelEntity> event) {

        event.async(
            register(event.getAfter())
        );
    }

    @EventListener
    public void handleEvent(EntityDeletedEvent<NotifySubscriberChannelEntity> event) {
        event.async(
            unregister(event.getEntity())
        );
    }

    @Subscribe(value = "/_sys/notify-channel/unregister", features = Subscription.Feature.broker)
    public void unregister(NotifySubscriberChannelEntity entity) {
        channels.remove(entity.getId());
    }

    @Subscribe(value = "/_sys/notify-channel/register", features = Subscription.Feature.broker)
    public Mono<Void> register(NotifySubscriberChannelEntity entity) {<FILL_FUNCTION_BODY>}

    private Mono<NotifyChannel> createChannel(NotifySubscriberChannelEntity entity) {
        NotifyChannelProvider provider = providers.get(entity.getChannelProvider());
        if (null == provider) {
            return Mono.empty();
        }
        return provider.createChannel(entity.getChannelConfiguration());
    }

    private Mono<Void> unregister(List<NotifySubscriberChannelEntity> entities) {
        for (NotifySubscriberChannelEntity entity : entities) {
            unregister(entity);
        }
        return Flux.fromIterable(entities)
                   .flatMap(e -> eventBus.publish("/_sys/notify-channel/unregister", e))
                   .then();
    }

    private Mono<Void> register(List<NotifySubscriberChannelEntity> entities) {
        return Flux.fromIterable(entities)
                   .flatMap(e -> register(e)
                       .then(eventBus.publish("/_sys/notify-channel/register", e)))
                   .then();

    }

    @Override
    public void run(String... args) throws Exception {
        channelRepository
            .createQuery()
            .where(NotifySubscriberChannelEntity::getState, NotifyChannelState.enabled)
            .fetch()
            .flatMap(e -> this
                .register(e)
                .onErrorResume(er -> {
                    log.warn("register notify channel error", er);
                    return Mono.empty();
                }))
            .subscribe();

    }
}

        if (entity.getState() == NotifyChannelState.disabled) {
            channels.remove(entity.getId());
        } else {
            return channels
                .compute(entity.getId(), (ignore, old) -> {
                    if (null != old) {
                        old.dispose();
                    }
                    return createChannel(entity);
                })
                .then();
        }
        return Mono.empty();
    
<no_super_class>