从Kafka消息中获取Token并注入所有Feign调用的实现方案

核心思路 #

Kafka消费逻辑与Feign调用默认处于同一线程上下文，因此可以通过ThreadLocal暂存从Kafka消息中获取的OAuth Token，再借助Feign拦截器统一将Token注入到所有Feign请求的请求头中。

具体实现步骤 #

1. 定义ThreadLocal工具类 #

用于安全存储和获取当前线程的OAuth Token，避免线程复用导致的Token污染：

public class OAuthTokenHolder {
    private static final ThreadLocal<String> TOKEN_HOLDER = new ThreadLocal<>();

    public static void setToken(String token) {
        TOKEN_HOLDER.set(token);
    }

    public static String getToken() {
        return TOKEN_HOLDER.get();
    }

    public static void clearToken() {
        TOKEN_HOLDER.remove();
    }
}

2. 在Kafka消费者中存储Token #

消费Kafka消息时，从消息体提取Token并存入ThreadLocal，务必在消费完成后清理Token：

@Component
public class KafkaMessageConsumer {

    @KafkaListener(topics = "your_topic", groupId = "your_consumer_group")
    public void consumeMessage(ConsumerRecord<String, BusinessMessageDto> record) {
        try {
            // 从消息DTO中取出传递的OAuth Token
            String oauthToken = record.value().getOauthToken();
            OAuthTokenHolder.setToken(oauthToken);
            
            // 执行业务逻辑，调用Feign客户端
            serviceCFeignClient.submitUpdateRequest(updateParam);
        } finally {
            // 强制清理ThreadLocal，防止线程复用导致Token泄露
            OAuthTokenHolder.clearToken();
        }
    }
}

3. 实现Feign请求拦截器 #

自定义拦截器，自动从ThreadLocal中获取Token并添加到Feign请求的Authorization头：

@Component
public class OAuthTokenFeignInterceptor implements RequestInterceptor {

    @Override
    public void apply(RequestTemplate template) {
        String token = OAuthTokenHolder.getToken();
        if (token != null) {
            // 按照OAuth2规范构造Authorization头
            template.header("Authorization", "Bearer " + token);
        }
    }
}

关键注意事项 #

• 线程安全保障：必须在finally块中清理ThreadLocal，因为Kafka消费者线程是复用的，残留Token会导致后续请求携带错误凭证。
• Token有效性校验：Service B需先验证Token的合法性（如过期时间、权限范围），避免将无效Token传递给Service C。
• 异步场景适配：若Service B使用异步线程处理业务（如@Async），ThreadLocal内容不会自动传递到异步线程，此时需手动传递Token，或改用InheritableThreadLocal（需注意线程池复用的风险）。

内容的提问来源于stack exchange，提问作者Ubaid Khanzada