Proxyless的多活流量和微服務(wù)治理

@Extensible("LoadBalancer")public interface LoadBalancer {
int ORDER_RANDOM_WEIGHT = 0;
int ORDER_ROUND_ROBIN = ORDER_RANDOM_WEIGHT + 1;
default <T extends Endpoint> T choose(List<T> endpoints, Invocation<?> invocation) {        Candidate<T> candidate = elect(endpoints, invocation);return candidate == null ? null : candidate.getTarget();    }
    <T extends Endpoint> Candidate<T> elect(List<T> endpoints, Invocation<?> invocation);}

@Extension(value = RoundRobinLoadBalancer.LOAD_BALANCER_NAME, order = LoadBalancer.ORDER_ROUND_ROBIN)@ConditionalOnProperties(value = {@ConditionalOnProperty(value = GovernanceConfig.CONFIG_LIVE_ENABLED, matchIfMissing = true),@ConditionalOnProperty(value = GovernanceConfig.CONFIG_LANE_ENABLED, matchIfMissing = true),@ConditionalOnProperty(value = GovernanceConfig.CONFIG_FLOW_CONTROL_ENABLED, matchIfMissing = true)}, relation = ConditionalRelation.OR)public class RoundRobinLoadBalancer extends AbstractLoadBalancer {
public static final String LOAD_BALANCER_NAME = "ROUND_ROBIN";
private static final Function<Long, AtomicLong> COUNTER_FUNC = s -> new AtomicLong(0L);
private final Map<Long, AtomicLong> counters = new ConcurrentHashMap<>();
private final AtomicLong global = new AtomicLong(0);
@Overridepublic <T extends Endpoint> Candidate<T> doElect(List<T> endpoints, Invocation<?> invocation) {        AtomicLong counter = global;        ServicePolicy servicePolicy = invocation.getServiceMetadata().getServicePolicy();        LoadBalancePolicy loadBalancePolicy = servicePolicy == null ? null : servicePolicy.getLoadBalancePolicy();if (loadBalancePolicy != null) {            counter = counters.computeIfAbsent(loadBalancePolicy.getId(), COUNTER_FUNC);        }long count = counter.getAndIncrement();if (count < 0) {            counter.set(0);            count = counter.getAndIncrement();        }// Ensure the index is within the bounds of the endpoints list.int index = (int) (count % endpoints.size());return new Candidate<>(endpoints.get(index), index);    }}

@Target(ElementType.TYPE)@Retention(RetentionPolicy.RUNTIME)@Documentedpublic @interface Injectable {boolean enable() default true;}

@Target(ElementType.FIELD)@Retention(RetentionPolicy.RUNTIME)@Documentedpublic @interface Inject {String value() default "";boolean nullable() default false;ResourcerType loader() default ResourcerType.CORE_IMPL;}

@Target(ElementType.TYPE)@Retention(RetentionPolicy.RUNTIME)@Documentedpublic @interface Configurable {String prefix() default "";boolean auto() default false;}

@Target(ElementType.FIELD)@Retention(RetentionPolicy.RUNTIME)@Documentedpublic @interface Config {String value() default "";boolean nullable() default true;}

@Injectable@Extension(value = "CircuitBreakerFilter", order = OutboundFilter.ORDER_CIRCUIT_BREAKER)public class CircuitBreakerFilter implements OutboundFilter, ExtensionInitializer {@Injectprivate Map<String, CircuitBreakerFactory> factories;
@Inject(nullable = true)private CircuitBreakerFactory defaultFactory;
@Inject(GovernanceConfig.COMPONENT_GOVERNANCE_CONFIG)private GovernanceConfig governanceConfig;
    ...}
@Configurable(prefix = "app")public class Application {@Setter@Config("name")private String name;
@Setter@Config("service")private AppService service;
    ...}

import net.bytebuddy.ByteBuddy;import net.bytebuddy.implementation.MethodDelegation;import net.bytebuddy.implementation.SuperMethodCall;import net.bytebuddy.matcher.ElementMatchers;
import java.lang.reflect.Method;
// 原始類class SimpleClass {public void sayHello() {System.out.println("Hello, World!");    }}
// 攔截器class SimpleInterceptor {public static void beforeMethod() {System.out.println("Before saying hello");    }
public static void afterMethod() {System.out.println("After saying hello");    }}
public class ByteBuddyExample {public static void main(String[] args) throws Exception {// 使用ByteBuddy創(chuàng)建增強(qiáng)類Class<?> dynamicType = new ByteBuddy()                .subclass(SimpleClass.class)                .method(ElementMatchers.named("sayHello"))                .intercept(MethodDelegation.to(SimpleInterceptor.class)                        .andThen(SuperMethodCall.INSTANCE))                .make()                .load(ByteBuddyExample.class.getClassLoader())                .getLoaded();
// 創(chuàng)建增強(qiáng)類的實(shí)例Object enhancedInstance = dynamicType.getDeclaredConstructor().newInstance();
// 調(diào)用增強(qiáng)后的方法Method sayHelloMethod = enhancedInstance.getClass().getMethod("sayHello");        sayHelloMethod.invoke(enhancedInstance);    }}

Before saying helloHello, World!After saying hello

.└── plugin    ├── dubbo    │   ├── joylive-registry-dubbo2.6-1.0.0.jar    │   ├── joylive-registry-dubbo2.7-1.0.0.jar    │   ├── joylive-registry-dubbo3-1.0.0.jar    │   ├── joylive-router-dubbo2.6-1.0.0.jar    │   ├── joylive-router-dubbo2.7-1.0.0.jar    │   ├── joylive-router-dubbo3-1.0.0.jar    │   ├── joylive-transmission-dubbo2.6-1.0.0.jar    │   ├── joylive-transmission-dubbo2.7-1.0.0.jar    │   └── joylive-transmission-dubbo3-1.0.0.jar

@Extensible("PluginDefinition")public interface PluginDefinition {
ElementMatcher<TypeDesc> getMatcher();
InterceptorDefinition[] getInterceptors();}

public interface InterceptorDefinition {
ElementMatcher<MethodDesc> getMatcher();
Interceptor getInterceptor();}

public interface Interceptor {        void onEnter(ExecutableContext ctx);
    void onSuccess(ExecutableContext ctx);
    void onError(ExecutableContext ctx);
    void onExit(ExecutableContext ctx);}

.└── 多活空間    ├── 單元路由變量(*)    ├── 單元(*)    │   ├── 分區(qū)(*)    ├── 單元規(guī)則(*)    ├── 多活域名(*)    │   ├── 單元子域名(*)    │   ├── 路徑(*)    │   │   ├── 業(yè)務(wù)參數(shù)(*)

[  {"apiVersion": "apaas.cos.com/v2alpha1","kind": "MultiLiveSpace","metadata": {"name": "mls-abcdefg1","namespace": "apaas-livespace"    },"spec": {"id": "v4bEh4kd6Jvu5QBX09qYq-qlbcs","code": "7Jei1Q5nlDbx0dRB4ZKd","name": "TestLiveSpace","version": "2023120609580935201","tenantId": "tenant1","units": [      ],"domains": [      ],"unitRules": [      ],"variables": [      ]    }  }]

.└── 服務(wù)    ├── 分組*    │   ├── 路徑*    │   │   ├── 方法*

@Injectable@Extension(value = "ServiceConfigDefinition_v3", order = PluginDefinition.ORDER_REGISTRY)@ConditionalOnProperties(value = {@ConditionalOnProperty(value = GovernanceConfig.CONFIG_LIVE_ENABLED, matchIfMissing = true),@ConditionalOnProperty(value = GovernanceConfig.CONFIG_LANE_ENABLED, matchIfMissing = true),@ConditionalOnProperty(value = GovernanceConfig.CONFIG_FLOW_CONTROL_ENABLED, matchIfMissing = true)}, relation = ConditionalRelation.OR) @ConditionalOnClass(ServiceConfigDefinition.TYPE_CONSUMER_CONTEXT_FILTER)@ConditionalOnClass(ServiceConfigDefinition.TYPE_SERVICE_CONFIG)public class ServiceConfigDefinition extends PluginDefinitionAdapter {
protected static final String TYPE_SERVICE_CONFIG = "org.apache.dubbo.config.ServiceConfig";
private static final String METHOD_BUILD_ATTRIBUTES = "buildAttributes";
private static final String[] ARGUMENT_BUILD_ATTRIBUTES = new String[]{"org.apache.dubbo.config.ProtocolConfig"    };
// ......
public ServiceConfigDefinition() {this.matcher = () -> MatcherBuilder.named(TYPE_SERVICE_CONFIG);this.interceptors = new InterceptorDefinition[]{new InterceptorDefinitionAdapter(                        MatcherBuilder.named(METHOD_BUILD_ATTRIBUTES).                                and(MatcherBuilder.arguments(ARGUMENT_BUILD_ATTRIBUTES)),                        () -> new ServiceConfigInterceptor(application, policySupplier))        };    }}

public class ServiceConfigInterceptor extends InterceptorAdaptor {
// ......
    @Override    public void onSuccess(ExecutableContext ctx) {        MethodContext methodContext = (MethodContext) ctx;
Map<String, String> map = (Map<String, String>) methodContext.getResult();        application.label(map::putIfAbsent);
// ......    }}

public class ServiceConfigInterceptor extends InterceptorAdaptor {
@Overridepublic void onSuccess(ExecutableContext ctx) {        MethodContext methodContext = (MethodContext) ctx;
// ......
        AbstractInterfaceConfig config = (AbstractInterfaceConfig) ctx.getTarget();        ApplicationConfig application = config.getApplication();        String registerMode = application.getRegisterMode();if (DEFAULT_REGISTER_MODE_INSTANCE.equals(registerMode)) {            policySupplier.subscribe(application.getName());        } else if (DEFAULT_REGISTER_MODE_INTERFACE.equals(registerMode)) {            policySupplier.subscribe(config.getInterface());        } else {            policySupplier.subscribe(application.getName());            policySupplier.subscribe(config.getInterface());        }    }}

@Injectable@Extension(value = "ClassLoaderFilterDefinition_v3")@ConditionalOnProperty(value = GovernanceConfig.CONFIG_LIVE_ENABLED, matchIfMissing = true)@ConditionalOnProperty(value = GovernanceConfig.CONFIG_LIVE_DUBBO_ENABLED, matchIfMissing = true)@ConditionalOnProperty(value = GovernanceConfig.CONFIG_REGISTRY_ENABLED, matchIfMissing = true)@ConditionalOnProperty(value = GovernanceConfig.CONFIG_TRANSMISSION_ENABLED, matchIfMissing = true)@ConditionalOnClass(ClassLoaderFilterDefinition.TYPE_CLASSLOADER_FILTER)public class ClassLoaderFilterDefinition extends PluginDefinitionAdapter {
protected static final String TYPE_CLASSLOADER_FILTER = "org.apache.dubbo.rpc.filter.ClassLoaderFilter";
private static final String METHOD_INVOKE = "invoke";
protected static final String[] ARGUMENT_INVOKE = new String[]{"org.apache.dubbo.rpc.Invoker","org.apache.dubbo.rpc.Invocation"    };
// ......
public ClassLoaderFilterDefinition() {this.matcher = () -> MatcherBuilder.named(TYPE_CLASSLOADER_FILTER);this.interceptors = new InterceptorDefinition[]{new InterceptorDefinitionAdapter(                        MatcherBuilder.named(METHOD_INVOKE).                                and(MatcherBuilder.arguments(ARGUMENT_INVOKE)),                        () -> new ClassLoaderFilterInterceptor(context)                )        };    }}

public class ClassLoaderFilterInterceptor extends InterceptorAdaptor {
private final InvocationContext context;
public ClassLoaderFilterInterceptor(InvocationContext context) {this.context = context;    }
@Overridepublic void onEnter(ExecutableContext ctx) {        MethodContext mc = (MethodContext) ctx;        Object[] arguments = mc.getArguments();        Invocation invocation = (Invocation) arguments[1];try {            context.inbound(new DubboInboundInvocation(new DubboInboundRequest(invocation), context));        } catch (RejectException e) {            Result result = new AppResponse(new RpcException(RpcException.FORBIDDEN_EXCEPTION, e.getMessage()));            mc.setResult(result);            mc.setSkip(true);        }    }}

public interface InvocationContext {
// ......
default <R extends InboundRequest> void inbound(InboundInvocation<R> invocation) {        InboundFilterChain.Chain chain = new InboundFilterChain.Chain(getInboundFilters());        chain.filter(invocation);    }
}

public class ClusterInterceptor extends InterceptorAdaptor {
// ......
    @Overridepublic void onEnter(ExecutableContext ctx) {        MethodContext mc = (MethodContext) ctx;        Object[] arguments = ctx.getArguments();        DubboCluster3 cluster = clusters.computeIfAbsent((AbstractClusterInvoker<?>) ctx.getTarget(), DubboCluster3::new);List<Invoker<?>> invokers = (List<Invoker<?>>) arguments[1];List<DubboEndpoint<?>> instances = invokers.stream().map(DubboEndpoint::of).collect(Collectors.toList());        DubboOutboundRequest request = new DubboOutboundRequest((Invocation) arguments[0]);        DubboOutboundInvocation invocation = new DubboOutboundInvocation(request, context);        DubboOutboundResponse response = cluster.request(context, invocation, instances);if (response.getThrowable() != null) {            mc.setThrowable(response.getThrowable());        } else {            mc.setResult(response.getResponse());        }        mc.setSkip(true);    }
}

apiVersion: apps/v1kind: Deploymentmetadata:  labels:    app: joylive-demo-springcloud2021-provider    x-live-enabled: "true"  name: joylive-demo-springcloud2021-providerspec:  replicas: 1  selector:    matchLabels:      app: joylive-demo-springcloud2021-provider  template:    metadata:      labels:        app: joylive-demo-springcloud2021-provider        x-live-enabled: "true"    spec:      containers:        - env:            - name: CONFIG_LIVE_SPACE_API_TYPE              value: multilive            - name: CONFIG_LIVE_SPACE_API_URL              value: http://api.live.local/v1            - name: CONFIG_LIVE_SPACE_API_HEADERS              value: pin=demo            - name: CONFIG_SERVICE_API_TYPE              value: jmsf            - name: CONFIG_SERVICE_API_URL              value: http://api.jmsf.local/v1            - name: LIVE_LOG_LEVEL              value: info            - name: CONFIG_LANE_ENABLED              value: "false"            - name: NACOS_ADDR              value: nacos-server.nacos.svc:8848            - name: NACOS_USERNAME              value: nacos            - name: NACOS_PASSWORD              value: nacos            - name: APPLICATION_NAME              value: springcloud2021-provider            - name: APPLICATION_SERVICE_NAME              value: service-provider            - name: APPLICATION_SERVICE_NAMESPACE              value: default            - name: SERVER_PORT              value: "18081"            - name: APPLICATION_LOCATION_REGION              value: region1            - name: APPLICATION_LOCATION_ZONE              value: zone1            - name: APPLICATION_LOCATION_LIVESPACE_ID              value: v4bEh4kd6Jvu5QBX09qYq-qlbcs            - name: APPLICATION_LOCATION_UNIT              value: unit1            - name: APPLICATION_LOCATION_CELL              value: cell1            - name: APPLICATION_LOCATION_LANESPACE_ID              value: "1"            - name: APPLICATION_LOCATION_LANE              value: production          image: hub-vpc.jdcloud.com/jmsf/joylive-demo-springcloud2021-provider:1.1.0-5aab82b3-AMD64          imagePullPolicy: Always          name: joylive-demo-springcloud2021-provider          ports:            - containerPort: 18081              name: http              protocol: TCP          resources:            requests:              cpu: "4"              memory: "8Gi"            limits:              cpu: "4"              memory: "8Gi"          terminationMessagePath: /dev/termination-log          terminationMessagePolicy: File      dnsPolicy: ClusterFirst      restartPolicy: Always      schedulerName: default-scheduler      securityContext: { }      terminationGracePeriodSeconds: 30