Soul源码阅读体验sofa代理【第四天】

sofa接入参考文档

https://dromara.org/zh-cn/docs/soul/user-dubbo.html

体验基础的sofa代理功能

1、启动 zookeeper ，默认端口2181
2、启动 soul-examples-sofa 下面的 TestSofaApplication
查看控制台，sofa的服务注册成功

{
    "appName": "sofa",
    "contextPath": "/sofa",
    "path": "/sofa/findAll",
    "pathDesc": "Get all data",
    "rpcType": "sofa",
    "serviceName": "org.dromara.soul.examples.dubbo.api.service.DubboTestService",
    "methodName": "findAll",
    "ruleName": "/sofa/findAll",
    "parameterTypes": "",
    "rpcExt": "{\"loadbalance\":\"hash\",\"retries\":3,\"timeout\":-1}",
    "enabled": true
}

看起来和Dubbo的没什么区别
3、admin控制台开启sofa插件支持

image.png

4、查看sofa注册的服务

image.png

5、sofa服务需要加 @SoulSofaClient 才能被识别注册
4、访问一下 http://localhost:9195/sofa/findById?id=1 试试
提示 can not match selector data: divide

运行不成功，发现加载了14个插件，但是唯独没有Sofa的插件，需要添加对应的jar包，soul-bootstrap 的 pom.xml 添加一下jar包

<dependency>
            <groupId>com.alipay.sofa</groupId>
            <artifactId>sofa-rpc-all</artifactId>
            <version>5.7.6</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-client</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-framework</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.apache.curator</groupId>
            <artifactId>curator-recipes</artifactId>
            <version>4.0.1</version>
        </dependency>
        <dependency>
            <groupId>org.dromara</groupId>
            <artifactId>soul-spring-boot-starter-plugin-sofa</artifactId>
            <version>${project.version}</version>
        </dependency>

MetaData(id=1350423582223273984, 
appName=sofa, 
contextPath=null, 
path=/sofa/findById, 
rpcType=sofa, serviceName=org.dromara.soul.examples.dubbo.api.service.DubboTestService, 
methodName=findById, 
parameterTypes=java.lang.String, 
rpcExt={"loadbalance":"hash","retries":3,"timeout":-1}, 
enabled=true)

再次访问

{
    "code": -103,
    "message": "Service invocation exception, or no result is returned!",
    "data": null
}

有点懵，再次把所有项目重启下
这次返回：

{"code":500,"message":"Internal Server Error"}

看一下报错

2021-01-17 01:41:30.227 ERROR 8909 --- [work-threads-10] o.d.soul.web.handler.GlobalErrorHandler  : [b7e443cc] Resolved [NullPointerException: null] for HTTP GET /sofa/findById
2021-01-17 01:41:30.228 ERROR 8909 --- [work-threads-10] a.w.r.e.AbstractErrorWebExceptionHandler : [b7e443cc]  500 Server Error for HTTP GET "/sofa/findById?id=1"

java.lang.NullPointerException: null
    at org.dromara.soul.plugin.sofa.proxy.SofaProxyService.genericInvoker(SofaProxyService.java:81) ~[classes/:na]
    Suppressed: reactor.core.publisher.FluxOnAssembly$OnAssemblyException: 
Error has been observed at the following site(s):
    |_ checkpoint ⇢ org.dromara.soul.web.configuration.ErrorHandlerConfiguration$1 [DefaultWebFilterChain]
    |_ checkpoint ⇢ org.dromara.soul.web.filter.WebSocketParamFilter [DefaultWebFilterChain]
    |_ checkpoint ⇢ org.dromara.soul.web.filter.FileSizeFilter [DefaultWebFilterChain]
    |_ checkpoint ⇢ org.dromara.soul.bootstrap.filter.HealthFilter [DefaultWebFilterChain]
    |_ checkpoint ⇢ org.springframework.boot.actuate.metrics.web.reactive.server.MetricsWebFilter [DefaultWebFilterChain]
    |_ checkpoint ⇢ HTTP GET "/sofa/findById?id=1" [ExceptionHandlingWebHandler]
Stack trace:
        at org.dromara.soul.plugin.sofa.proxy.SofaProxyService.genericInvoker(SofaProxyService.java:81) ~[classes/:na]
        at org.dromara.soul.plugin.sofa.SofaPlugin.doExecute(SofaPlugin.java:78) ~[classes/:na]
        at org.dromara.soul.plugin.base.AbstractSoulPlugin.execute(AbstractSoulPlugin.java:97) ~[classes/:na]
        at org.dromara.soul.web.handler.SoulWebHandler$DefaultSoulPluginChain.lambda$execute$0(SoulWebHandler.java:107) ~[classes/:na]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:44) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoDefer.subscribe(MonoDefer.java:52) [reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.Mono.subscribe(Mono.java:4105) ~[reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.publisher.MonoSubscribeOn$SubscribeOnSubscriber.run(MonoSubscribeOn.java:124) ~[reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84) ~[reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37) ~[reactor-core-3.3.1.RELEASE.jar:3.3.1.RELEASE]
        at java.util.concurrent.FutureTask.run$$$capture(FutureTask.java:266) ~[na:1.8.0_211]
        at java.util.concurrent.FutureTask.run(FutureTask.java) ~[na:1.8.0_211]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180) ~[na:1.8.0_211]
        at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293) ~[na:1.8.0_211]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[na:1.8.0_211]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[na:1.8.0_211]
        at java.lang.Thread.run(Thread.java:748) ~[na:1.8.0_211]

喜闻乐见的空指针异常哈，
SofaProxyService 中的
private final SofaParamResolveService sofaParamResolveService;
为空，我们看一下是哪里关联的
SofaProxyService 中的 genericInvoker方法进行的调用

public Mono<Object> genericInvoker(final String body, final MetaData metaData, final ServerWebExchange exchange) throws SoulException {
        ConsumerConfig<GenericService> reference = ApplicationConfigCache.getInstance().get(metaData.getPath());
        if (Objects.isNull(reference) || StringUtils.isEmpty(reference.getInterfaceId())) {
            ApplicationConfigCache.getInstance().invalidate(metaData.getServiceName());
            reference = ApplicationConfigCache.getInstance().initRef(metaData);
        }
        GenericService genericService = reference.refer();
        Pair<String[], Object[]> pair;
        if (null == body || "".equals(body) || "{}".equals(body) || "null".equals(body)) {
            pair = new ImmutablePair<>(new String[]{}, new Object[]{});
        } else {
            # 这里调用，空指针了
            pair = sofaParamResolveService.buildParameter(body, metaData.getParameterTypes());
        }

查看了下，官网上面说的是下面的内容

参数传递：
- 通过 http post 方式访问网关，通过body，json类型传递。
- 更多参数类型传递，可以参考 soul-test-sofa 中的接口定义，以及参数传递方式。这个已经不能访问了
单个java bean参数类型（默认）这个是哪里显示默认实现的？默认实现调用有参数的调用不通
自定义实现多参数支持：
- 在你搭建的网关项目中，新增一个类 A，实现 org.dromara.soul.plugin.api.sofa.SofaParamResolveService。

   public interface SofaParamResolveService {

       /**
        * Build parameter pair.
        * this is Resolve http body to get sofa param.
        *
        * @param body           the body
        * @param parameterTypes the parameter types
        * @return the pair
        */
       Pair<String[], Object[]> buildParameter(String body, String parameterTypes);
   }

body为http中body传的json字符串。
parameterTypes: 匹配到的方法参数类型列表，如果有多个,则使用,分割。
Pair中，left为参数类型，right为参数值，这是sofa泛化调用的标准。
把你的类注册成Spring的bean，覆盖默认的实现。

   @Bean
    public SofaParamResolveService A() {
            return new A();
    }

发现是参数解析的问题后，尝试调用下 http://localhost:9195/sofa/findAll，成功返回

{
    "code": 200,
    "message": "Access to success!",
    "data": {
        "name": "hello world Soul Sofa , findAll",
        "id": "26369681"
    }
}

新增 SofaParamResolveServiceImpl 实现下SofaParamResolveService的接口，下面的代码是Dubbo泛化调用的，为何Sofa这个没有提供默认的实现？

@Slf4j
@Component
public class SofaParamResolveServiceImpl implements SofaParamResolveService {
    @Override
    public Pair<String[], Object[]> buildParameter(final String body, final String parameterTypes) {
        Map<String, Object> paramMap = GsonUtils.getInstance().toObjectMap(body);
        String[] parameter = StringUtils.split(parameterTypes, ",");
        if (parameter.length == 1 && !isBaseType(parameter[0])) {
            for (String key : paramMap.keySet()) {
                Object obj = paramMap.get(key);
                if (obj instanceof JsonObject) {
                    paramMap.put(key, GsonUtils.getInstance().convertToMap(obj.toString()));
                } else if (obj instanceof JsonArray) {
                    paramMap.put(key, GsonUtils.getInstance().fromList(obj.toString(), Object.class));
                } else {
                    paramMap.put(key, obj);
                }
            }
            return new ImmutablePair<>(parameter, new Object[]{paramMap});
        }
        List<Object> list = new LinkedList<>();
        for (String key : paramMap.keySet()) {
            Object obj = paramMap.get(key);
            if (obj instanceof JsonObject) {
                list.add(GsonUtils.getInstance().convertToMap(obj.toString()));
            } else if (obj instanceof JsonArray) {
                list.add(GsonUtils.getInstance().fromList(obj.toString(), Object.class));
            } else {
                list.add(obj);
            }
        }
        Object[] objects = list.toArray();
        return new ImmutablePair<>(parameter, objects);
    }

    private boolean isBaseType(final String paramType) {
        return paramType.startsWith("java") || paramType.startsWith("[Ljava");
    }
}

调用成功，结果如下

{
    "code": 200,
    "message": "Access to success!",
    "data": {
        "name": "hello world Soul Sofa, findById",
        "id": "1"
    }
}

经过这么一搞，整体流程我更加清楚了些，新增需要之后代码需要阅读的点
1、Rpc的泛化调用，基于 org.apache.commons.lang3.tuple.Pair
2、Sofa大概了解下
3、RPC调用是使用元数据，保存了基础的调用RPC服务的数据；与其他调用有什么异同点？
4、如何封装jar，加载了jar包就实现插件加载

Soul源码阅读 体验sofa代理【第四天】

sofa接入参考文档

体验基础的sofa代理功能

查看了下，官网上面说的是下面的内容

推荐阅读更多精彩内容

Soul源码阅读体验sofa代理【第四天】