本文介绍 HSF 如何进行异步调用。

同步调用

HSF 的 IO 操作是异步操作,客户端同步调用的本质是执行 future.get(timeout)操作,等待服务端的结果返回,这里的 timeout 就是客户端生效的超时时间 (默认 3000ms)。同步调用时序图如下所示。SAE服务HSF应用开发异步调用

对于客户端来说,并不是所有的 HSF 服务都是需要同步等待服务端返回结果的,对于这些服务,HSF 提供异步调用的形式,让客户端不必同步阻塞在 HSF 操作上。 异步调用在发起调用时,HSF 服务的调用结果都是返回值的默认值,例如返回类型是 int,则会返回 0,返回类型是 Object,则会返回 null。而真正的结果,是在 HSFResponseFuture 或者回调函数(callback)中获得的。

Future 异步调用

HSF 发起调用后,用户可以在上下文中获取跟返回结果关联的 HSFFuture 对象,获取对象后调用 HSFFuture.getResponse(timeout)获取服务端的返回结果。Future 异步调用的时序图如下所示。

SAE应用HSF应用开发之Future异步调用
  • API 形式配置 HSF 服务

    HSF 提供了方法级别的异步调用配置,格式为 name:${methodName};type:future,由于只用方法名字来标识方法,所以并不区分重载的方法。同名的方法都会被设置为同样的调用方式。

    HSFApiConsumerBean hsfApiConsumerBean = new HSFApiConsumerBean();
    hsfApiConsumerBean.setInterfaceName("com.alibaba.middleware.hsf.guide.api.service.OrderService");
    hsfApiConsumerBean.setVersion("1.0.0");
    hsfApiConsumerBean.setGroup("HSF");
    // [设置] 异步 future 调用
    List<String> asyncallMethods = new ArrayList<String>();
    // 格式:name:{methodName};type:future
    asyncallMethods.add("name:queryOrder;type:future");
    hsfApiConsumerBean.setAsyncallMethods(asyncallMethods);
    
    hsfApiConsumerBean.init(true);
    
    // [代理] 获取 HSF 代理
    OrderService orderService = (OrderService) hsfApiConsumerBean.getObject();
    // ---------------------- 调用 -----------------------//
    // [调用] 发起 HSF 异步调用, 返回 null
    OrderModel orderModel = orderService.queryOrder(1L);
    // 及时在当前调用上下文中,获取 future 对象;因为该对象是放在 `ThreadLocal` 中,同一线程中后续调用会覆盖 future 对象,所以要及时取出。
    HSFFuture hsfFuture = HSFResponseFuture.getFuture();
    
    // do something else
    
    // 这里才真正地获取结果,如果调用还未完成,将阻塞等待结果,5000ms 是等待结果的最大时间
    try {
        System.out.println(hsfFuture.getResponse(5000));
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

    HSF 默认的超时配置是 3000ms,如果过了超时时间,业务对象未返回,这时调用 HSFFuture.getResponse 会抛出超时异常;HSFFuture.getResponse(timeout),如果这里的 timeout 时间内,业务结果没有返回,也没有超时,可以调用多次执行 getResponse 去获取结果。

  • Spring 配置 HSF 服务

    Spring 框架在应用中广泛使用,如果不想以 API 的形式配置 HSF 服务,您可以使用 Spring XML 的形式进行配置,上述例子中 API 配置等同于如下 XML 配置。

    <bean id="orderService" class="com.taobao.hsf.app.spring.util.HSFSpringConsumerBean">
        <property name="interfaceName" value="com.alibaba.middleware.hsf.guide.api.service.OrderService"/>
        <property name="version" value="1.0.0"/>
        <property name="group" value="HSF"/>
        <!--[设置] 订阅服务的接口 -->
        <property name="asyncallMethods">
            <list>
                <value>name:queryOrder;type:future</value>
            </list>
         </property>
    </bean>
  • 注解配置 HSF 服务

    SpringBoot 广泛使用的今天,使用注解装配 SpringBean 也成为一种选择,HSF 也支持使用注解进行配置,用来订阅服务。

    在项目中增加依赖 starter。

    <dependency>
        <groupId>com.alibaba.boot</groupId>
        <artifactId>pandora-hsf-spring-boot-starter</artifactId>
    </dependency>

    通常一个 HSF Consumer 会在多个地方使用,但并不需要在每次使用的地方都用 @HSFConsumer 来标记。只需要写一个统一个 Config 类,然后在其它需要使用的地方,直接 @Autowired 注入即可上述例子中的 API 配置等同于如下注解配置。

    @Configuration
    public class HsfConfig {
        @HSFConsumer(serviceVersion = "1.0.0", serviceGroup = "HSF", futureMethods = "sayHelloInFuture")
        OrderService orderService;
    
    }

    在使用时直接注入即可。

    @Autowired
    OrderService orderService;

Callback 异步调用

客户端配置为 callback 方式调用时,需要配置一个实现了 HSFResponseCallback 接口的 listener,结果返回之后,HSF 会调用 HSFResponseCallback 中的方法。时序图如下所示。

Callback 异步调用
  • API 形式配置 HSF 服务

    callback 的调用上下文

    用户在调用前还可以通过 CallbackInvocationContext.setContext(Object obj),来设置一个关于本次调用的上下文信息,该信息存放在 threadlocal 中。在 listener 的回调函数中,可以通过 CallbackInvocationContext.getContext() 来获取该对象

    回调函数示例

    public class CallbackHandler implements HSFResponseCallback {
    
        // 业务异常时会触发
        @Override
        public void onAppException(Throwable t) {
            t.printStackTrace();
        }
    
        // 业务返回结果
        @Override
        public void onAppResponse(Object result) {
            // 取 callback 调用时设置的上下文
            Object context = CallbackInvocationContext.getContext();
    
            System.out.println(result.toString() + context);
        }
        //HSF 异常
        @Override
        public void onHSFException(HSFException e) {
            e.printStackTrace();
        }
    }

    接口 callback 方法配置

    HSFApiConsumerBean hsfApiConsumerBean = new HSFApiConsumerBean();
    hsfApiConsumerBean.setInterfaceName("com.alibaba.middleware.hsf.guide.api.service.OrderService");
    hsfApiConsumerBean.setVersion("1.0.0");
    hsfApiConsumerBean.setGroup("HSF");
    // [设置] 异步 callback 调用
    List<String> asyncallMethods = new ArrayList<String>();
    asyncallMethods.add("name:queryOrder;type:callback;listener:com.alibaba.middleware.hsf.CallbackHandler");
    hsfApiConsumerBean.setAsyncallMethods(asyncallMethods);
    hsfApiConsumerBean.init(true);
    // [代理] 获取 HSF 代理
    OrderService orderService = (OrderService) hsfApiConsumerBean.getObject();
     // 可选步骤,设置上下文。CallbackHandler 中通过 api 可以获取到
    CallbackInvocationContext.setContext("in callback");
    // 发起调用
    orderService.queryOrder(1L); // 这里返回的其实是 null
    // 清理上下文
    CallbackInvocationContext.setContext(null);
    // do something else

    在调用线程中可以设置上下文,然后在 listener 中获取使用。相对于 Future 异步调用,callback 会立即知晓结果的返回。

  • Spring 配置 HSF 服务
    <bean id="CallHelloWorld" class="com.taobao.hsf.app.spring.util.HSFSpringConsumerBean">
        <!--[设置] 订阅服务的接口 -->
        <property name="interfaceName" value="com.alibaba.middleware.hsf.guide.api.service.OrderService"/>
        <!--[设置] 服务的版本 -->
        <property name="version" value="1.0.0"/>
        <!--[设置] 服务的归组 -->
        <property name="group" value="HSF"/>
        <property name="asyncallMethods">
            <list>
                <!--future 的含义为通过 Future 的方式去获取请求执行的结果,例如先调用下远程的接口,接着在同一线程继续做别的事情,然后再在同一线程中通过 Future 来获取结果 -->
                <!--name:methodName;type:future|callback-->
                <value>name:queryOrder;type:callback;listener:com.alibaba.middleware.hsf.CallbackHandler</value>
            </list>
         </property>
    </bean>
  • 注解配置 HSF 接口方法为 callback 调用
    @AsyncOn(interfaceName = OrderService.class, methodName = "queryOrder")
    public class CallbackHandler implements HSFResponseCallback {
    
        @Override
        public void onAppException(Throwable t) {
            t.printStackTrace();
        }
    
        @Override
        public void onAppResponse(Object result) {
            // 取 callback 调用时设置的上下文
            Object context = CallbackInvocationContext.getContext();
    
            System.out.println(result.toString() + context);
        }
    
        @Override
        public void onHSFException(HSFException e) {
            e.printStackTrace();
        }
    
    
    }
    注意 回调函数是由单独的线程池( LinkedBlockingQueue 无限队列)来调用的,不要做太费时间的操作,避免影响其他请求的 onAppResponse 回调。 callback 线程默认的 corePoolSize, maxPoolSize 是机器 cpu 数目。 下面的 -D 参数可以去自定义配置。
    • CALLBACK 线程池最小配置: -Dhsf.callback.min.poolsize
    • CALLBACK 线程池最小配置: -Dhsf.callback.min.poolsize