dubbo之callback

考虑这样的场景，当消费者调用服务提供者后，服务提供者需要执行一段时间，比如一分钟左右，才能有结果，这时就有2种方案，一种是同步方案，即等着，第二种是异步方案，有结果之后让服务提供者回调消费者的一个接口，显然同步的方案肯定是不行的，线程资源可能在很短的时间内就会耗尽，因此第二种明显是最优方案，针对这种场景，dubbo提供了callback机制，下面我们来一起看下。

public interface MyCallbackService {
    void addListener(String key, CallbackListener listener);
}

消费者实现该接口提供回调实现类，提供者调用回调实现类执行回调操作。

public interface CallbackListener {
    void changed(String msg);
}

public class CallbackServiceImpl implements MyCallbackService {
     
    private final Map listeners = new ConcurrentHashMap();
  
    public CallbackServiceImpl() {
        // 定时触发变更
        Thread t = new Thread(new Runnable() {
            public void run() {
                while(true) {
                    try {
                        for(Map.Entry entry : listeners.entrySet()){
                           try {
                               entry.getValue().changed(getChanged(entry.getKey()));
                           } catch (Throwable t) {
                               listeners.remove(entry.getKey());
                           }
                        }
                        Thread.sleep(5000); // 定时触发变更通知
                    } catch (Throwable t) { // 防御容错
                        t.printStackTrace();
                    }
                }
            }
        });
        t.setDaemon(true);
        t.start();
    }
  
    public void addListener(String key, CallbackListener listener) {
        listeners.put(key, listener);
        // 添加监听器直接触发一次变更
        listener.changed(getChanged(key)); // 发送变更通知
    }
     
    private String getChanged(String key) {
        return "Changed: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
    }
}

public class ProviderWithCallbackMain {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext ac
                = new ClassPathXmlApplicationContext("meta-INF/spring/provider-with-call.xml");
        ac.start();
        System.in.read(); // 按任意键退出
    }
}

public class ConsumerWithCallbackMain {
    public static void main(String[] args) throws Exception {
        ClassPathXmlApplicationContext context
                = new ClassPathXmlApplicationContext("consumer-with-callback.xml");
        context.start();
        MyCallbackService callbackService = (MyCallbackService) context.getBean("callbackService");

        callbackService.addListener("foo.bar", new CallbackListener(){
            public void changed(String msg) {
                System.out.println("callback1:" + msg);
            }
        });

        System.in.read();
    }
}

假定服务提供者已经启动成功。

...
callback1:Changed: 2021-11-30 13:39:58
callback1:Changed: 2021-11-30 13:39:58
callback1:Changed: 2021-11-30 13:40:03
callback1:Changed: 2021-11-30 13:40:08

可以看到每5秒钟就会被调用一次。