class DefaultDimensionUserService extends GenericReactiveCrudService<DimensionUserEntity, String> {

    @Autowired
    private ApplicationEventPublisher eventPublisher;

    //处理用户被删除时,同步删除维度绑定信息
    @EventListener
    public void handleUserDeleteEntity(UserDeletedEvent event) {
        event.async(this.createDelete()
                        .where(DimensionUserEntity::getUserId, event.getUser().getId())
                        .execute()
                        .doOnSuccess(i -> log.debug("user deleted,clear user dimension!"))
        );
    }

    //转发保存维度信息到DimensionBindEvent事件,并清空权限缓存
    @EventListener
    public void dispatchDimensionBind(EntitySavedEvent<DimensionUserEntity> event) {
        event.async(
                this.publishEvent(Flux.fromIterable(event.getEntity()), DimensionBindEvent::new)
        );
    }

    //新增绑定时转发DimensionBindEvent并清空用户权限信息
    @EventListener
    public void dispatchDimensionBind(EntityCreatedEvent<DimensionUserEntity> event) {
        event.async(
                this.publishEvent(Flux.fromIterable(event.getEntity()), DimensionBindEvent::new)
        );
    }

    //删除绑定时转发DimensionUnbindEvent并清空用户权限信息
    @EventListener
    public void dispatchDimensionUnbind(EntityDeletedEvent<DimensionUserEntity> event) {
        event.async(
                this.publishEvent(Flux.fromIterable(event.getEntity()), DimensionUnbindEvent::new)
        );
    }

    //修改绑定信息时清空权限
    @EventListener
    public void handleModifyEvent(EntityModifyEvent<DimensionUserEntity> event) {
        event.async(
                this.clearUserCache(event.getAfter())
        );
    }

    //维度被删除时同时删除绑定信息
    @EventListener
    public void handleDimensionDeletedEntity(EntityDeletedEvent<DimensionEntity> event) {
        event.async(
                Flux.fromIterable(event.getEntity())
                    .collect(groupingBy(DimensionEntity::getTypeId,
                                        mapping(DimensionEntity::getId, toSet())))
                    .flatMapIterable(Map::entrySet)
                    .flatMap(entry -> this
                            .createDelete()
                            .where(DimensionUserEntity::getDimensionTypeId, entry.getKey())
                            .in(DimensionUserEntity::getDimensionId, entry.getValue())
                            .execute())
        );

    }

    private Flux<DimensionUserEntity> publishEvent(Publisher<DimensionUserEntity> stream,
                                                   Function3<String, String, List<String>, AsyncEvent> event) {<FILL_FUNCTION_BODY>}

    private Mono<Void> clearUserCache(List<DimensionUserEntity> entities) {
        return Flux.fromIterable(entities)
                   .map(DimensionUserEntity::getUserId)
                   .distinct()
                   .collectList()
                   .flatMap(list -> ClearUserAuthorizationCacheEvent.of(list).publish(eventPublisher));
    }

}

        Flux<DimensionUserEntity> cache = Flux.from(stream).doOnNext(DimensionUserEntity::generateId).cache();

        Set<Mono<Void>> jobs = ConcurrentHashMap.newKeySet();

        return cache
                .groupBy(DimensionUserEntity::getDimensionTypeId)
                .flatMap(typeGroup -> {
                    String type = typeGroup.key();
                    return typeGroup
                            .groupBy(DimensionUserEntity::getDimensionId)
                            .flatMap(dimensionIdGroup -> {
                                String dimensionId = dimensionIdGroup.key();
                                return dimensionIdGroup
                                        .map(DimensionUserEntity::getUserId)
                                        .collectList()
                                        .doOnNext(userIdList -> jobs.add(event.apply(type, dimensionId, userIdList).publish(eventPublisher)))
                                        .flatMapIterable(Function.identity());
                            });
                })
                .<Collection<String>>collect(HashSet::new, Collection::add)
                .flatMap(userList -> ClearUserAuthorizationCacheEvent.of(userList).publish(eventPublisher))
                .then(Mono.defer(() -> Flux.concat(jobs).then()))
                .thenMany(cache);
    
<methods>public non-sealed void <init>() ,public ReactiveRepository<org.hswebframework.web.system.authorization.api.entity.DimensionUserEntity,java.lang.String> getRepository() <variables>private ReactiveRepository<org.hswebframework.web.system.authorization.api.entity.DimensionUserEntity,java.lang.String> repository