dubbo参数回调机制-CFANZ编程社区

概念

参数回调就是dubbo的provider调用consumer代码.
参数回调分为同步回调和异步回调

同步回调官网文档地址:https://dubbo.apache.org/zh/docs/v2.7/user/examples/callback-parameter/ 异步回调官网文档地址
https://dubbo.apache.org/zh/docs/v2.7/user/examples/async-call/ 异步回调就是利用了Java的CompletableFuture

gitee代码地址

https://gitee.com/zjj19941/ZJJ_Dubbo.git 的callback项目就是代码,自己准备个zookeeper,然后改一下配置文件,先执行provider,再执行consumer,即可看到效果

使用

interface项目声明一个回调接口类

package com.zjj;
// 回调参数
public interface DemoServiceListener {
    String changed(String msg) throws InterruptedException;
}

interface项目接口添加一个回调方法

package com.zjj;


public interface DemoService {

    // 添加回调逻辑,不然的话消费者无法进行回调
    //参数2是key, 不同的key的回调函数可以不一样. 这个key就类似于
    default String sayHello(String name,  DemoServiceListener listener) throws InterruptedException {
        return null;
    };
}

provider编写回调方法

public String sayHello(String name, DemoServiceListener callback) 是回调方法

@Service也配置了一些参数回调相关的参数

package com.zjj.provider.service;

import com.zjj.DemoService;
import com.zjj.DemoServiceListener;
import org.apache.dubbo.config.annotation.Argument;
import org.apache.dubbo.config.annotation.Method;
import org.apache.dubbo.config.annotation.Service;


// DemoService的sayHello方法的index=1的参数是回调对象，服务消费者可以调用addListener方法来添加回调对象，服务提供者一旦执行回调对象的方法就会通知给服务消费者
//arguments = {@Argument(index = 1, callback = true)})} 的意思是 sayHello方法的第二个参数是回调参数
//callbacks = 3 的意思是这个服务最多支持多少个回调
@Service(version = "default", timeout =1000000  ,  methods = {@Method(name = "sayHello", arguments = {@Argument(index = 1, callback = true)})}, callbacks = 3)
public class CallBackDemoService implements DemoService {



    /**
     * 回调方法
     */
    @Override
    public String sayHello(String name,  DemoServiceListener callback) throws InterruptedException {
        System.out.println("1.开始执行回调服务,参数name是:" + name + "   毫秒值是 :" + System.currentTimeMillis());

        //在执行回调函数
        String rs = callback.changed("我是provider传过去的入参");
        Thread.sleep(1000);


        System.out.println("3.回调的结果是: " + rs  + "   毫秒值是 :" + System.currentTimeMillis());
        return String.format( name);  // 正常访问
    }

}

consumer编写回调类

这个回调类的逻辑会被provider调用

package com.zjj;

//参数回调
public class DemoServiceListenerImpl implements DemoServiceListener {

    @Override
    public String changed(String msg) throws InterruptedException {

        System.out.println("2.被回调了："+msg  + "   毫秒值是 :" + System.currentTimeMillis());
        Thread.sleep(1000);
        return "回调结果";
    }
}

consumer方法访问provider

参数2就是指定回调方法.指定的这个回调方法会被provider调用到.

String rs = demoService.sayHello(“周瑜”, new DemoServiceListenerImpl());

package com.zjj;

import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.ConfigurableApplicationContext;

@SpringBootApplication
public class DubboConsumerDemo  {

  //@Reference注解就是用于标记这个服务具体使用了生产者的哪个接口实现
    @Reference(version = "default",timeout = 100000000)
    private DemoService demoService;

    public static void main(String[] args) throws InterruptedException {
        ConfigurableApplicationContext context = SpringApplication.run(DubboConsumerDemo.class);
        DemoService demoService = context.getBean(DemoService.class);
        //等provider调用完了回调之后,这个方法才会给返回值结果.
        String rs = demoService.sayHello("周瑜",  new DemoServiceListenerImpl());
        System.out.println("4.rs:" + rs  + "   毫秒值是 :" + System.currentTimeMillis());
    }

}

执行结果

provider控制台

1.开始执行回调服务,参数name是:周瑜   毫秒值是 :1639306614182
3.回调的结果是: 回调结果   毫秒值是 :1639306616205

consumer控制台

2.被回调了：我是provider传过去的入参   毫秒值是 :1639306614201
4.rs:周瑜   毫秒值是 :1639306616210

看控制台打印的毫秒值就能看出来方法执行顺序.

1.consumer先调用provider的sayHello方法,
2.provider调用consumer的回调方法
3.provider收到回调结果后方法执行完毕
4.consumer的sayHello执行完毕,获取到方法的返回结果

上面是同步回调的处理顺序