Open Feign架构 - 远程通信的利器
太叔弘壮
前言
Feign 是一个声明式的 web service 客户端。Spring Cloud 中的 Open Feign 在 Feign 的基础上支持 Spring MVC 注解、JAX-RS 注解，同时集成了 Ribbon、Eureka（对于负载均衡，同样支持 Spring Cloud LoadBalancer）
对 @EnableFeignClients 注解进行解析，扫描指定路径下的所有标注 @FeignClient 注解的接口，解析其标注的 @FeignClient 注解，然后生成动态代理。最后组装成一个 Request 对象，进行远程通信。
代码样例
方式一 、服务提供者模块定义 Feign Client 接口
order-api模块
@RequestMapping("order")
public interface OrderService {
@GetMapping("orders")
String orders();
}
@FeignClient("spring-cloud-order")
public interface OrderServiceFeignClient extends OrderService {
}
order-application模块
@RestController
public class OrderServiceImpl implements OrderService {
@Value("${server.port}")
private Integer port;
@Override
public String orders() {
return String.valueOf(port);
}
}
server:
port: 8081
spring:
application:
name: spring-cloud-order
server:
port: 8082
spring:
application:
name: spring-cloud-order
user-service模块
@RestController
public class UserController {
@Autowired
private OrderServiceFeignClient orderServiceFeignClient;
@GetMapping("user2")
public String user2() {
return orderServiceFeignClient.orders();
}
}
@SpringBootApplication
@EnableFeignClients(basePackages = {
"com.mzs.order.api.clients"
})
public class SpringCloudUserApplication {
public static void main(String[] args) {
SpringApplication.run(SpringCloudUserApplication.class, args);
}
}
方式二 、在服务消费者模块定义 Feign Client 接口
order-api模块
public interface OrderService {
String orders();
}
order-service模块
@RestController
@RequestMapping("order")
public class OrderController {
@Autowired
private OrderService orderService;
@GetMapping("orders")
public String orders() {
return orderService.orders();
}
}
server:
port: 8081
spring:
application:
name: spring-cloud-order
server:
port: 8082
spring:
application:
name: spring-cloud-order
user-service模块
@FeignClient("spring-cloud-order")
@RequestMapping("order")
public interface OrderServiceFeignClient {
@GetMapping("orders")
String orders();
}
spring-cloud-order:
ribbon:
listOfServers: localhost:8081,localhost:8082
@RestController
public class UserController {
@Autowired
private OrderServiceFeignClient orderServiceFeignClient;
@GetMapping("user2")
public String user2() {
return orderServiceFeignClient.orders();
}
}
@SpringBootApplication
@EnableFeignClients
public class SpringCloudUserApplication {
public static void main(String[] args) {
SpringApplication.run(SpringCloudUserApplication.class, args);
}
}
源码分析
一、@EnableFeignClients
用于扫描所有标注 @FeignClient注解的接口。
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
@Documented
@Import(FeignClientsRegistrar.class)
public @interface EnableFeignClients {
......
}
二、FeignClientsRegistrar
FeignClientsRegistrar 实现了 ImportBeanDefinitionRegistrar 接口，看下 registerBeanDefitions(…) 方法的具体实现，如下：
@Override
public void registerBeanDefinitions(AnnotationMetadata metadata,
BeanDefinitionRegistry registry) {
// 注册 @EnableFeignClients 注解 defaultConfiguration 属性指定的配置类
registerDefaultConfiguration(metadata, registry);
// 获取 @FeignClient 注解的属性以及属性值，填充到 FeignClientFactoryBean
// 从而注册到 IOC 容器中
registerFeignClients(metadata, registry);
}
1、registerDefaultConfiguration(…) 方法
private void registerDefaultConfiguration(AnnotationMetadata metadata,
BeanDefinitionRegistry registry) {
// 获取 @EnableFeignClients 注解的属性以及属性值
Map<String, Object> defaultAttrs = metadata
.getAnnotationAttributes(EnableFeignClients.class.getName(), true);
// 如果 @EnableFeignClients 注解有指定 defaultConfiguration 属性
if (defaultAttrs != null && defaultAttrs.containsKey("defaultConfiguration")) {
String name;
if (metadata.hasEnclosingClass()) {
name = "default." + metadata.getEnclosingClassName();
}
else {
name = "default." + metadata.getClassName();
}
// 注册指定的配置类
registerClientConfiguration(registry, name,
defaultAttrs.get("defaultConfiguration"));
}
}
private void registerClientConfiguration(BeanDefinitionRegistry registry, Object name,
Object configuration) {
// 创建关于FeignClientSpecification的BeanDefinitionBuilder实例
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(FeignClientSpecification.class);
builder.addConstructorArgValue(name);
// 将@EnableFeignClients 注解 defaultConfiguration 属性指定的配置类封装到 FeignClientSpecification
builder.addConstructorArgValue(configuration);
// 注册到 IOC 容器中
registry.registerBeanDefinition(
name + "." + FeignClientSpecification.class.getSimpleName(),
builder.getBeanDefinition());
}
2、registerFeignClients(…) 方法
public void registerFeignClients(AnnotationMetadata metadata,
BeanDefinitionRegistry registry) {
LinkedHashSet<BeanDefinition> candidateComponents = new LinkedHashSet<>();
// 获取 @EnableFeignClients 注解的属性以及属性值
Map<String, Object> attrs = metadata
.getAnnotationAttributes(EnableFeignClients.class.getName());
AnnotationTypeFilter annotationTypeFilter = new AnnotationTypeFilter(
FeignClient.class);
final Class<?>[] clients = attrs == null ? null
: (Class<?>[]) attrs.get("clients");
// 如果 @EnableFeignClients 注解没有指定 clients 属性
// 则扫描指定包路径下所有标注 @FeignClient 注解的类然后放到 candidateComponents 集合中
if (clients == null || clients.length == 0) {
ClassPathScanningCandidateComponentProvider scanner = getScanner();
scanner.setResourceLoader(this.resourceLoader);
scanner.addIncludeFilter(new AnnotationTypeFilter(FeignClient.class));
Set<String> basePackages = getBasePackages(metadata);
for (String basePackage : basePackages) {
candidateComponents.addAll(scanner.findCandidateComponents(basePackage));
}
}
// 如果 @EnableFeignClients 注解有指定 clients 属性
// 则加载这些标注 @FeignClient 注解的类然后放到 candidateComponents 集合中
else {
for (Class<?> clazz : clients) {
candidateComponents.add(new AnnotatedGenericBeanDefinition(clazz));
}
}
for (BeanDefinition candidateComponent : candidateComponents) {
if (candidateComponent instanceof AnnotatedBeanDefinition) {
AnnotatedBeanDefinition beanDefinition = (AnnotatedBeanDefinition) candidateComponent;
AnnotationMetadata annotationMetadata = beanDefinition.getMetadata();
Assert.isTrue(annotationMetadata.isInterface(),
"@FeignClient can only be specified on an interface");
// 获取 @FeignClient 注解的属性以及属性值
Map<String, Object> attributes = annotationMetadata
.getAnnotationAttributes(FeignClient.class.getCanonicalName());
// 注册 @FeignClient 注解 configuration 属性指定的配置类
String name = getClientName(attributes);
// 将@FeignClient 注解 configuration 属性指定的配置类封装到 FeignClientSpecification，从而注册到 IOC 容器中
registerClientConfiguration(registry, name,
attributes.get("configuration"));
// 获取 @FeignClient 注解的属性以及属性值，填充到 FeignClientFactoryBean，从而注册到 IOC 容器中。
registerFeignClient(registry, annotationMetadata, attributes);
}
}
}
registerClientConfiguration方法
private void registerClientConfiguration(BeanDefinitionRegistry registry, Object name,
Object configuration) {
BeanDefinitionBuilder builder = BeanDefinitionBuilder
.genericBeanDefinition(FeignClientSpecification.class);
builder.addConstructorArgValue(name);
builder.addConstructorArgValue(configuration);
registry.registerBeanDefinition(
name + "." + FeignClientSpecification.class.getSimpleName(),
builder.getBeanDefinition());
}
registerFeignClient方法
private void registerFeignClient(BeanDefinitionRegistry registry,
AnnotationMetadata annotationMetadata, Map<String, Object> attributes) {
String className = annotationMetadata.getClassName();
BeanDefinitionBuilder definition = BeanDefinitionBuilder
.genericBeanDefinition(FeignClientFactoryBean.class);
validate(attributes);
definition.addPropertyValue("url", getUrl(attributes));
definition.addPropertyValue("path", getPath(attributes));
String name = getName(attributes);
definition.addPropertyValue("name", name);
String contextId = getContextId(attributes);
definition.addPropertyValue("contextId", contextId);
definition.addPropertyValue("type", className);
definition.addPropertyValue("decode404", attributes.get("decode404"));
definition.addPropertyValue("fallback", attributes.get("fallback"));
definition.addPropertyValue("fallbackFactory", attributes.get("fallbackFactory"));
definition.setAutowireMode(AbstractBeanDefinition.AUTOWIRE_BY_TYPE);
String alias = contextId + "FeignClient";
AbstractBeanDefinition beanDefinition = definition.getBeanDefinition();
beanDefinition.setAttribute(FactoryBean.OBJECT_TYPE_ATTRIBUTE, className);
boolean primary = (Boolean) attributes.get("primary");
beanDefinition.setPrimary(primary);
String qualifier = getQualifier(attributes);
if (StringUtils.hasText(qualifier)) {
alias = qualifier;
}
BeanDefinitionHolder holder = new BeanDefinitionHolder(beanDefinition, className,
new String[] { alias });
BeanDefinitionReaderUtils.registerBeanDefinition(holder, registry);
}
三、FeignClientFactoryBean
FeignClientFactoryBean是FactoryBean接口的实现类。
@Override
public Object getObject() throws Exception {
return getTarget();
}
<T> T getTarget() {
FeignContext context = applicationContext.getBean(FeignContext.class);
// 使用FeignClientsConfiguration配置类以及配置文件中指定的相关bean组装Feign.Builder
Feign.Builder builder = feign(context);
if (!StringUtils.hasText(url)) {
if (!name.startsWith("http")) {
url = "http://" + name;
}
else {
url = name;
}
url += cleanPath();
return (T) loadBalance(builder, context,
new HardCodedTarget<>(type, name, url));
}
if (StringUtils.hasText(url) && !url.startsWith("http")) {
url = "http://" + url;
}
String url = this.url + cleanPath();
Client client = getOptional(context, Client.class);
if (client != null) {
if (client instanceof LoadBalancerFeignClient) {
client = ((LoadBalancerFeignClient) client).getDelegate();
}
if (client instanceof FeignBlockingLoadBalancerClient) {
client = ((FeignBlockingLoadBalancerClient) client).getDelegate();
}
builder.client(client);
}
Targeter targeter = get(context, Targeter.class);
return (T) targeter.target(this, builder, context,
new HardCodedTarget<>(type, name, url));
}
protected <T> T loadBalance(Feign.Builder builder, FeignContext context,
HardCodedTarget<T> target) {
Client client = getOptional(context, Client.class);
if (client != null) {
builder.client(client);
// 从上下文获取DefaultTargeter实例
Targeter targeter = get(context, Targeter.class);
// 执行DefaultTargeter#target方法获取代理对象
return targeter.target(this, builder, context, target);
}
throw new IllegalStateException(
"No Feign Client for loadBalancing defined. Did you forget to include spring-cloud-starter-netflix-ribbon?");
}
四、DefaultTargeter
FeignAutoConfiguration自动装配类中定义 DefaultTargeter 类型的bean
@Configuration(proxyBeanMethods = false)
@ConditionalOnMissingClass("feign.hystrix.HystrixFeign")
protected static class DefaultFeignTargeterConfiguration {
@Bean
@ConditionalOnMissingBean
public Targeter feignTargeter() {
return new DefaultTargeter();
}
}
class DefaultTargeter implements Targeter {
@Override
public <T> T target(FeignClientFactoryBean factory, Feign.Builder feign,
FeignContext context, Target.HardCodedTarget<T> target) {
// 生成代理对象
return feign.target(target);
}
}
五、Feign
public <T> T target(Target<T> target) {
// 生成代理对象
return build().newInstance(target);
}
public Feign build() {
Client client = Capability.enrich(this.client, capabilities);
Retryer retryer = Capability.enrich(this.retryer, capabilities);
List<RequestInterceptor> requestInterceptors = this.requestInterceptors.stream()
.map(ri -> Capability.enrich(ri, capabilities))
.collect(Collectors.toList());
Logger logger = Capability.enrich(this.logger, capabilities);
Contract contract = Capability.enrich(this.contract, capabilities);
Options options = Capability.enrich(this.options, capabilities);
Encoder encoder = Capability.enrich(this.encoder, capabilities);
Decoder decoder = Capability.enrich(this.decoder, capabilities);
InvocationHandlerFactory invocationHandlerFactory =
Capability.enrich(this.invocationHandlerFactory, capabilities);
QueryMapEncoder queryMapEncoder = Capability.enrich(this.queryMapEncoder, capabilities);
SynchronousMethodHandler.Factory synchronousMethodHandlerFactory =
new SynchronousMethodHandler.Factory(client, retryer, requestInterceptors, logger,
logLevel, decode404, closeAfterDecode, propagationPolicy, forceDecoding);
ParseHandlersByName handlersByName =
new ParseHandlersByName(contract, options, encoder, decoder, queryMapEncoder,
errorDecoder, synchronousMethodHandlerFactory);
return new ReflectiveFeign(handlersByName, invocationHandlerFactory, queryMapEncoder);
}
六、ReflectiveFeign
ReflectiveFeign 在 Feign#build 方法中进行实例化。
@Override
public <T> T newInstance(Target<T> target) {
// 通过ParseHandlersByName#apply方法，获取Target实例的Map<方法描述名，SynchronousMethodHandler>
Map<String, MethodHandler> nameToHandler = targetToHandlersByName.apply(target);
Map<Method, MethodHandler> methodToHandler = new LinkedHashMap<Method, MethodHandler>();
List<DefaultMethodHandler> defaultMethodHandlers = new LinkedList<DefaultMethodHandler>();
// 遍历目标Class对象的所有方法
for (Method method : target.type().getMethods()) {
if (method.getDeclaringClass() == Object.class) {
continue;
// 如果方法有default关键字修饰
} else if (Util.isDefault(method)) {
// 构建DefaultMethodHandler实例
DefaultMethodHandler handler = new DefaultMethodHandler(method);
defaultMethodHandlers.add(handler);
// 将方法以及对应的DefaultMethodHandler实例添加到methodToHandler
methodToHandler.put(method, handler);
} else {
// 在nameToHandler中查找指定方法描述名对应的SynchronousMethodHandler实例，然后将两者添加到methodToHandler
methodToHandler.put(method, nameToHandler.get(Feign.configKey(target.type(), method)));
}
}
// 调用InvocationHandlerFactory.Default#create方法创建 FeignInvocationHandler 实例
InvocationHandler handler = factory.create(target, methodToHandler);
// 使用JDK动态代理生成一个代理对象
T proxy = (T) Proxy.newProxyInstance(target.type().getClassLoader(),
new Class<?>[] {target.type()}, handler);
for (DefaultMethodHandler defaultMethodHandler : defaultMethodHandlers) {
defaultMethodHandler.bindTo(proxy);
}
// 返回该代理对象
return proxy;
}
ParseHandlersByName#apply
public Map<String, MethodHandler> apply(Target target) {
// 由SpringMvcContract将目标Class对象解析为方法元数据列表
List<MethodMetadata> metadata = contract.parseAndValidateMetadata(target.type());
Map<String, MethodHandler> result = new LinkedHashMap<String, MethodHandler>();
// 遍历方法元数据列表
for (MethodMetadata md : metadata) {
BuildTemplateByResolvingArgs buildTemplate;
// 创建不同类型的BuildTemplateByResolvingArgs实例
if (!md.formParams().isEmpty() && md.template().bodyTemplate() == null) {
buildTemplate =
new BuildFormEncodedTemplateFromArgs(md, encoder, queryMapEncoder, target);
} else if (md.bodyIndex() != null) {
buildTemplate = new BuildEncodedTemplateFromArgs(md, encoder, queryMapEncoder, target);
} else {
buildTemplate = new BuildTemplateByResolvingArgs(md, queryMapEncoder, target);
}
if (md.isIgnored()) {
result.put(md.configKey(), args -> {
throw new IllegalStateException(md.configKey() + " is not a method handled by feign");
});
} else {
// 由 SynchronousMethodHandler.Factory#create 方法创建一个SynchronousMethodHandler实例
// 然后将方法的描述名+SynchronousMethodHandler实例添加到result中
result.put(md.configKey(),
factory.create(target, md, buildTemplate, options, decoder, errorDecoder));
}
}
return result;
}
}
七、SpringMvcContract
@Override
public MethodMetadata parseAndValidateMetadata(Class<?> targetType, Method method) {
// 将方法的描述名，方法加入到processedMethods缓存中
processedMethods.put(Feign.configKey(targetType, method), method);
// 解析类、方法、参数标注的注解并进行相关处理
MethodMetadata md = super.parseAndValidateMetadata(targetType, method);
// 获取方法中标注的@RequestMapping注解
RequestMapping classAnnotation = findMergedAnnotation(targetType,
RequestMapping.class);
// 如果方法有标注@RequestMapping注解
if (classAnnotation != null) {
if (!md.template().headers().containsKey(ACCEPT)) {
// 解析方法中@RequestMapping注解的produces属性，添加到方法元数据中
parseProduces(md, method, classAnnotation);
}
if (!md.template().headers().containsKey(CONTENT_TYPE)) {
// 解析方法中@RequestMapping注解的consumes属性，添加到方法元数据中
parseConsumes(md, method, classAnnotation);
}
// 解析方法中@RequestMapping注解的headers属性，添加到方法元数据中
parseHeaders(md, method, classAnnotation);
}
// 返回方法元数据
return md;
}
八、FeignInvocationHandler
FeignInvocationHandler 由 InvocationHandlerFactory.Default#create 方法进行实例化。
static final class Default implements InvocationHandlerFactory {
@Override
public InvocationHandler create(Target target, Map<Method, MethodHandler> dispatch) {
return new ReflectiveFeign.FeignInvocationHandler(target, dispatch);
}
}
InvocationHandler接口的实现类，重点看下它的 invoker 方法的实现，如下：
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
if ("equals".equals(method.getName())) {
try {
Object otherHandler =
args.length > 0 && args[0] != null ? Proxy.getInvocationHandler(args[0]) : null;
return equals(otherHandler);
} catch (IllegalArgumentException e) {
return false;
}
} else if ("hashCode".equals(method.getName())) {
return hashCode();
} else if ("toString".equals(method.getName())) {
return toString();
}
// 找到方法对应的MethodHandler，然后执行 invoke 方法
return dispatch.get(method).invoke(args);
}
九、SynchronousMethodHandler
@Override
public Object invoke(Object[] argv) throws Throwable {
// 创建RequestTemplate实例
RequestTemplate template = buildTemplateFromArgs.create(argv);
Options options = findOptions(argv);
Retryer retryer = this.retryer.clone();
while (true) {
try {
// 核心方法 executeAndDecode
return executeAndDecode(template, options);
} catch (RetryableException e) {
try {
retryer.continueOrPropagate(e);
} catch (RetryableException th) {
Throwable cause = th.getCause();
if (propagationPolicy == UNWRAP && cause != null) {
throw cause;
} else {
throw th;
}
}
if (logLevel != Logger.Level.NONE) {
logger.logRetry(metadata.configKey(), logLevel);
}
continue;
}
}
}
Object executeAndDecode(RequestTemplate template, Options options) throws Throwable {
// 组装Request对象
Request request = targetRequest(template);
if (logLevel != Logger.Level.NONE) {
logger.logRequest(metadata.configKey(), logLevel, request);
}
Response response;
long start = System.nanoTime();
try {
// 发送请求 - LoadBalancerFeignClient#execute(...)
response = client.execute(request, options);
response = response.toBuilder()
.request(request)
.requestTemplate(template)
.build();
} catch (IOException e) {
if (logLevel != Logger.Level.NONE) {
logger.logIOException(metadata.configKey(), logLevel, e, elapsedTime(start));
}
throw errorExecuting(request, e);
}
long elapsedTime = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start);
if (decoder != null)
// 对响应进行解码
return decoder.decode(response, metadata.returnType());
CompletableFuture<Object> resultFuture = new CompletableFuture<>();
asyncResponseHandler.handleResponse(resultFuture, metadata.configKey(), response,
metadata.returnType(),
elapsedTime);
try {
if (!resultFuture.isDone())
throw new IllegalStateException("Response handling not done");
return resultFuture.join();
} catch (CompletionException e) {
Throwable cause = e.getCause();
if (cause != null)
throw cause;
throw e;
}
}
十、LoadBalancerFeignClient
主要看下Ribbon在Open Feign中如何基于负载均衡实现请求的分发
@Override
public Response execute(Request request, Request.Options options) throws IOException {
try {
URI asUri = URI.create(request.url());
String clientName = asUri.getHost();
URI uriWithoutHost = cleanUrl(request.url(), clientName);
// this.delegate：DefaultFeignLoadBalancedConfiguration 配置类中设置的 Client.Default 实例
// requestConfig：RibbonClientConfiguration 配置类中设置的 DefaultClientConfigImpl 实例
FeignLoadBalancer.RibbonRequest ribbonRequest = new FeignLoadBalancer.RibbonRequest(
this.delegate, request, uriWithoutHost);
IClientConfig requestConfig = getClientConfig(options, clientName);
// 发送请求
return lbClient(clientName)
.executeWithLoadBalancer(ribbonRequest, requestConfig).toResponse();
}
catch (ClientException e) {
IOException io = findIOException(e);
if (io != null) {
throw io;
}
throw new RuntimeException(e);
}
}
public T executeWithLoadBalancer(final S request, final IClientConfig requestConfig) throws ClientException {
LoadBalancerCommand<T> command = buildLoadBalancerCommand(request, requestConfig);
try {
// 调用 LoadBalancerCommand#submit(…) 提交任务（看下该方法里面的 selectServer() 方法）
return command.submit(
new ServerOperation<T>() {
@Override
public Observable<T> call(Server server) {
URI finalUri = reconstructURIWithServer(server, request.getUri());
S requestForServer = (S) request.replaceUri(finalUri);
try {
return Observable.just(AbstractLoadBalancerAwareClient.this.execute(requestForServer, requestConfig));
}
catch (Exception e) {
return Observable.error(e);
}
}
})
.toBlocking()
.single();
} catch (Exception e) {
Throwable t = e.getCause();
if (t instanceof ClientException) {
throw (ClientException) t;
} else {
throw new ClientException(e);
}
}
}
LoadBalancerCommand#selectServer
private Observable<Server> selectServer() {
return Observable.create(new OnSubscribe<Server>() {
@Override
public void call(Subscriber<? super Server> next) {
try {
Server server = loadBalancerContext.getServerFromLoadBalancer(loadBalancerURI, loadBalancerKey);
next.onNext(server);
next.onCompleted();
} catch (Exception e) {
next.onError(e);
}
}
});
}
LoadBalancerContext#getServerFromLoadBalancer 方法截取关键代码
if (lb != null){
// 调用 ZoneAwareLoadBalancer#chooseServer(…) 基于负载均衡算法实现请求的分发
Server svc = lb.chooseServer(loadBalancerKey);
if (svc == null){
throw new ClientException(ClientException.ErrorType.GENERAL,
"Load balancer does not have available server for client: "
+ clientName);
}
host = svc.getHost();
if (host == null){
throw new ClientException(ClientException.ErrorType.GENERAL,
"Invalid Server for :" + svc);
}
logger.debug("{} using LB returned Server: {} for request {}", new Object[]{clientName, svc, original});
return svc;
}
类似资料：