public <T> Flux<GeoWithin<T>> georadiusbymember(
            Object key,
            Object member,
            double distanceMeters,
            GeoArgs geoArgs) {
        return Flux.defer(() -> {
            ByteBuf keyBuffer = serializationContext.encodeGeoKey(key);
            ByteBuf memberBuffer = serializationContext.encodeGeoMember(member);
            Flux<GeoWithin<T>> flux = commands
                    .createDissolvingFlux(() -> commandBuilder.georadiusbymember(GEORADIUSBYMEMBER,
                            keyBuffer,
                            memberBuffer,
                            distanceMeters,
                            GeoArgs.Unit.m.name(),
                            geoArgs));
            return flux.onErrorResume(RedisCommandExecutionException.class, e -> {
                String message = e.getMessage();
                if (message != null && message.endsWith("could not decode requested zset member")) {
                    return Flux.empty();
                }
                return Flux.error(e);
            })
                    .doFinally(signal -> {
                        ReferenceCountUtil.ensureReleased(keyBuffer);
                        ReferenceCountUtil.ensureReleased(memberBuffer);
                    });
        });
    }
ConnectionBuilder$PlainChannelInitializer ChannelGroupListener CommandEncoder
RedisHandshakeHandler CommandHandler ConnectionEventTrigger
ConnectionWatchdog

@see io.lettuce.core.SslConnectionBuilder.SslChannelInitializer
@see io.lettuce.core.SslConnectionBuilder.PlainChannelInitializer
turms-server-common/src/main/java/im/turms/server/common/storage/redis/TurmsRedisClient.java