如何在OIDC客户端链式调用Smallrye Uni<T>实现无阻塞认证

DevOps 小满

2026-7-28

问题描述

我正尝试为应用添加离线令牌（API）访问功能，需要接收Authorization: API $REFRESH_TOKEN请求头，通过OIDC客户端将其转换为Bearer令牌，再用于Keycloak授权服务。

目前遇到的核心问题：

在authenticate方法中，如何在不阻塞的情况下处理OIDC客户端返回的Uni<Tokens>，并返回Uni<SecurityIdentity>？当前实现靠2秒休眠才能运行，这明显不合理。
能否通过链式调用或订阅的方式，将Uni<SecurityIdentity>和Uni<Tokens>结合起来？
有没有办法更新oidc.authenticate(context, identityProviderManager)所用的RoutingContext，使其能响应授权头的更新？

代码实现

@Alternative
@Priority(1)
@ApplicationScoped
public class DexAuthenticationMechanism implements HttpAuthenticationMechanism {

    @Inject
    OidcAuthenticationMechanism oidc;

    @Inject
    OidcClients oidcClients;

    ExecutorService executor = Executors.newFixedThreadPool(10, r -> {
        return new Thread(r, "CUSTOM_TASK_EXECUTION_THREAD");
    });


    @Override
    public Uni<SecurityIdentity> authenticate(RoutingContext context, IdentityProviderManager identityProviderManager) {
        String authHeader = context.request().headers().get(HttpHeaders.AUTHORIZATION);
        String[] authSplit = authHeader.split("\\s+");
        if (authSplit.length == 2) {
            if ("BEARER".equalsIgnoreCase(authSplit[0])) {
                return oidc.authenticate(context, identityProviderManager);
            } else if ("BASIC".equalsIgnoreCase(authSplit[0])) {
                String accessToken = "";
                OidcClient basic = oidcClients.getClient("basic");
                context.request().headers().set(HttpHeaders.AUTHORIZATION, "Bearer "+accessToken);
                oidc.authenticate(context, identityProviderManager);
            } else if ("API".equalsIgnoreCase(authSplit[0])) {
                OidcClient api = oidcClients.getClient("api");
                Uni<Tokens> tokensUni = api.refreshTokens(authSplit[1]);
                tokensUni.runSubscriptionOn(executor)
                        .subscribe().with(
                            item -> context.request().headers().set(HttpHeaders.AUTHORIZATION, "Bearer " + item.getAccessToken())
                            , fail -> fail.printStackTrace()
                        );
                System.out.println(context.request().headers().get(HttpHeaders.AUTHORIZATION));
                //DON'T DO THIS
                try {
                    Thread.sleep(2000);
                }catch (Exception e){}
                System.out.println("After invoke");
                System.out.println(context.request().headers().get(HttpHeaders.AUTHORIZATION));
                return oidc.authenticate(context, identityProviderManager);
            } else {
                System.out.println("Cheese and crackers");
            }
        } else{
            System.out.println("Double cheese and crackers");
        }
        return Uni.createFrom().failure(new AuthenticationFailedException());
    }
}

配置文件

#quarkus.oidc-client.default
quarkus.oidc-client.enabled=true
quarkus.oidc-client.auth-server-url=${keycloakAuthServerUrl}/realms/${keycloakRealm}
quarkus.oidc-client.discovery-enabled=true
quarkus.oidc-client.client-id=${keycloakResource}
quarkus.oidc-client.credentials.secret=${keycloakCredentialsSecret}
quarkus.oidc-client.grant.type=client
###quarkus.oidc-client.proxy.host=http://${squidProxyHost}
###quarkus.oidc-client.proxy.port=${squidProxyPort}
#end quarkus.oidc-client.default

#quarkus.oidc-client."api"
quarkus.oidc-client."api".enabled=true
quarkus.oidc-client."api".auth-server-url=${keycloakAuthServerUrl}/realms/${keycloakRealm}
quarkus.oidc-client."api".discovery-enabled=true
quarkus.oidc-client."api".client-id=${keycloakResource}
quarkus.oidc-client."api".credentials.secret=${keycloakCredentialsSecret}
quarkus.oidc-client."api".grant.type=refresh
###quarkus.oidc-client.proxy.host=http://${squidProxyHost}
###quarkus.oidc-client.proxy.port=${squidProxyPort}
#end quarkus.oidc-client."api"

解决方案

1. 用响应式链式调用替代阻塞休眠

Quarkus的响应式模型中，不能手动订阅Uni后阻塞等待，必须通过链式操作编排流程。修改API类型授权头的处理逻辑，将令牌刷新、请求头更新、认证步骤串联：

else if ("API".equalsIgnoreCase(authSplit[0])) {
    OidcClient api = oidcClients.getClient("api");
    // 链式调用：先刷新令牌，再更新请求头，最后执行认证
    return api.refreshTokens(authSplit[1])
            .onItem().invoke(tokens -> {
                // 令牌刷新成功后更新Authorization头
                context.request().headers().set(HttpHeaders.AUTHORIZATION, "Bearer " + tokens.getAccessToken());
            })
            .onFailure().transform(throwable -> {
                // 统一转换认证失败异常
                return new AuthenticationFailedException("API令牌刷新失败", throwable);
            })
            // 令牌处理完成后，调用oidc认证方法
            .chain(ignored -> oidc.authenticate(context, identityProviderManager));
}

2. 移除自定义线程池

OidcClient本身是响应式实现，Quarkus会自动管理响应式线程，无需手动创建ExecutorService，直接删除相关代码即可。

3. 修复BASIC分支的逻辑漏洞

原代码中BASIC分支没有返回值，会导致后续执行默认失败逻辑，同样用响应式方式补全：

else if ("BASIC".equalsIgnoreCase(authSplit[0])) {
    OidcClient basic = oidcClients.getClient("basic");
    // 替换为实际的BASIC认证令牌获取逻辑（比如通过账号密码获取令牌）
    return basic.getTokens()
            .onItem().invoke(tokens -> 
                context.request().headers().set(HttpHeaders.AUTHORIZATION, "Bearer " + tokens.getAccessToken())
            )
            .onFailure().transform(throwable -> 
                new AuthenticationFailedException("BASIC认证失败", throwable)
            )
            .chain(ignored -> oidc.authenticate(context, identityProviderManager));
}

4. RoutingContext更新的说明

在响应式链式调用中，onItem().invoke()里对RoutingContext的修改会在令牌刷新成功后立即执行，后续的oidc.authenticate会自动使用更新后的请求头。链式操作保证了流程的串行执行，无需额外处理"订阅授权头更新"的问题。

修改后的完整代码

@Alternative
@Priority(1)
@ApplicationScoped
public class DexAuthenticationMechanism implements HttpAuthenticationMechanism {

    @Inject
    OidcAuthenticationMechanism oidc;

    @Inject
    OidcClients oidcClients;

    @Override
    public Uni<SecurityIdentity> authenticate(RoutingContext context, IdentityProviderManager identityProviderManager) {
        String authHeader = context.request().headers().get(HttpHeaders.AUTHORIZATION);
        if (authHeader == null) {
            return Uni.createFrom().failure(new AuthenticationFailedException("缺少Authorization请求头"));
        }
        
        String[] authSplit = authHeader.split("\\s+");
        if (authSplit.length == 2) {
            if ("BEARER".equalsIgnoreCase(authSplit[0])) {
                return oidc.authenticate(context, identityProviderManager);
            } else if ("BASIC".equalsIgnoreCase(authSplit[0])) {
                OidcClient basic = oidcClients.getClient("basic");
                // 替换为实际的BASIC认证令牌获取逻辑
                return basic.getTokens()
                        .onItem().invoke(tokens -> 
                            context.request().headers().set(HttpHeaders.AUTHORIZATION, "Bearer " + tokens.getAccessToken())
                        )
                        .onFailure().transform(throwable -> 
                            new AuthenticationFailedException("BASIC认证失败", throwable)
                        )
                        .chain(ignored -> oidc.authenticate(context, identityProviderManager));
            } else if ("API".equalsIgnoreCase(authSplit[0])) {
                OidcClient api = oidcClients.getClient("api");
                return api.refreshTokens(authSplit[1])
                        .onItem().invoke(tokens -> 
                            context.request().headers().set(HttpHeaders.AUTHORIZATION, "Bearer " + tokens.getAccessToken())
                        )
                        .onFailure().transform(throwable -> 
                            new AuthenticationFailedException("API令牌刷新失败", throwable)
                        )
                        .chain(ignored -> oidc.authenticate(context, identityProviderManager));
            } else {
                return Uni.createFrom().failure(new AuthenticationFailedException("不支持的授权类型"));
            }
        } else {
            return Uni.createFrom().failure(new AuthenticationFailedException("Authorization请求头格式无效"));
        }
    }
}

关键说明

响应式链式调用：通过onItem().invoke()处理成功逻辑，onFailure()处理异常，chain()衔接后续操作，全程非阻塞。
避免手动订阅：直接返回完整的Uni链，由Quarkus框架负责订阅和执行，无需手动调用subscribe()。
统一异常处理：将所有错误转换为AuthenticationFailedException，符合认证机制的规范。

内容的提问来源于stack exchange，提问作者Anthony Heaney

如何在OIDC客户端链式调用Smallrye Uni<T>实现无阻塞认证

代码实现 #

配置文件 #

1. 用响应式链式调用替代阻塞休眠 #

2. 移除自定义线程池 #

3. 修复BASIC分支的逻辑漏洞 #

4. RoutingContext更新的说明 #

修改后的完整代码 #

关键说明 #