09-30 09:01 阅读 102

Dubbo的基本使用与高级应用

本篇是Dubbo的使用篇，为后续的源码分析打基础。

1. 最简单的使用

开篇用一个最简单的例子，来介绍如何用Dubbo搭建一个简单的例子。

本例包括:

注册中心(用Zookeeper代替)
服务提供者(provider)
服务消费者(consumer)
接口(interface)

1.1 项目结构

创建项目，并创建三个Module，项目结构如下:

项目结构

1.2 项目依赖

整体pom.xml为:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>org.example</groupId>
    <artifactId>Study</artifactId>
    <packaging>pom</packaging>
    <version>1.0-SNAPSHOT</version>
    <modules>
        <module>consumer</module>
        <module>provider</module>
        <module>interface</module>
    </modules>

    <properties>
        <spring-boot.version>2.3.1.RELEASE</spring-boot.version>
        <dubbo.version>2.7.5</dubbo.version>
    </properties>

    <dependencyManagement>
        <dependencies>
            <!-- Spring Boot -->
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-dependencies</artifactId>
                <version>${spring-boot.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

            <!-- Apache Dubbo  -->
            <dependency>
                <groupId>org.apache.dubbo</groupId>
                <artifactId>dubbo-dependencies-bom</artifactId>
                <version>${dubbo.version}</version>
                <type>pom</type>
                <scope>import</scope>
            </dependency>

            <dependency>
                <groupId>org.apache.dubbo</groupId>
                <artifactId>dubbo</artifactId>
                <version>${dubbo.version}</version>
                <exclusions>
                    <exclusion>
                        <groupId>org.springframework</groupId>
                        <artifactId>spring</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>javax.servlet</groupId>
                        <artifactId>servlet-api</artifactId>
                    </exclusion>
                    <exclusion>
                        <groupId>log4j</groupId>
                        <artifactId>log4j</artifactId>
                    </exclusion>
                </exclusions>
            </dependency>
        </dependencies>
    </dependencyManagement>

</project>

consumer的依赖为:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>Study</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>consumer</artifactId>
    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

    <dependencies>

        <dependency>
            <groupId>org.example</groupId>
            <artifactId>interface</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <!-- Dubbo Spring Boot Starter -->
        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo-spring-boot-starter</artifactId>
            <version>2.7.5</version>
        </dependency>

        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <!-- Zookeeper dependencies -->
        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo-dependencies-zookeeper</artifactId>
            <version>${dubbo.version}</version>
            <type>pom</type>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo-rpc-http</artifactId>
            <version>${dubbo.version}</version>
        </dependency>

    </dependencies>


</project>

interface的依赖为:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>Study</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>interface</artifactId>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

provider的依赖为:

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <parent>
        <artifactId>Study</artifactId>
        <groupId>org.example</groupId>
        <version>1.0-SNAPSHOT</version>
    </parent>
    <modelVersion>4.0.0</modelVersion>

    <artifactId>provider</artifactId>

    <build>
        <plugins>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <configuration>
                    <source>8</source>
                    <target>8</target>
                </configuration>
            </plugin>
        </plugins>
    </build>

    <dependencies>
        <dependency>
            <groupId>org.example</groupId>
            <artifactId>interface</artifactId>
            <version>1.0-SNAPSHOT</version>
        </dependency>

        <!-- Dubbo Spring Boot Starter -->
        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo-spring-boot-starter</artifactId>
            <version>2.7.5</version>
        </dependency>

        <dependency>
            <groupId>org.jboss.resteasy</groupId>
            <artifactId>resteasy-jaxrs</artifactId>
            <version>3.0.19.Final</version>
        </dependency>

        <dependency>
            <groupId>javax.validation</groupId>
            <artifactId>validation-api</artifactId>
            <version>1.1.0.Final</version>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-web</artifactId>
        </dependency>

        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo</artifactId>
        </dependency>

        <!-- Zookeeper dependencies -->
        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo-dependencies-zookeeper</artifactId>
            <version>${dubbo.version}</version>
            <type>pom</type>
            <exclusions>
                <exclusion>
                    <groupId>org.slf4j</groupId>
                    <artifactId>slf4j-log4j12</artifactId>
                </exclusion>
            </exclusions>
        </dependency>


        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo-rpc-http</artifactId>
            <version>${dubbo.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.dubbo</groupId>
            <artifactId>dubbo-metadata-report-zookeeper</artifactId>
            <version>${dubbo.version}</version>
        </dependency>
    </dependencies>

</project>

1.3 具体代码

1.3.1 服务提供者

@Service注解暴露服务

实现类

package com.zyz.provider.service;

import com.zyz.DemoService;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.annotation.Service;
import org.apache.dubbo.rpc.RpcContext;

@Service(version = "default")
public class DefaultDemoService implements DemoService {

    @Override
    public String sayHello(String name) {
        System.out.println("执行了服务" + name);

        URL url = RpcContext.getContext().getUrl();
        return String.format("%s：%s, Hello, %s", url.getProtocol(), url.getPort(), name);  // 正常访问
    }
}

启动类

package com.zyz;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class DubboProviderDemo {

    public static void main(String[] args) {
        SpringApplication.run(DubboProviderDemo.class,args);
    }

}

配置文件application.properties:

# Spring boot application
spring.application.name=dubbo-provider-demo
server.port=8081

# Base packages to scan Dubbo Component: @org.apache.dubbo.config.annotation.Service
dubbo.scan.base-packages=com.zyz.provider.service
dubbo.application.name=${spring.application.name}


## Dubbo Registry
dubbo.registry.address=zookeeper://127.0.0.1:2181

# Dubbo Protocol
dubbo.protocol.name=dubbo
dubbo.protocol.port=20880

日志文件log4j.properties:

###set log levels###
log4j.rootLogger=info, stdout
###output to the console###
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.Target=System.out
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=[%d{dd/MM/yy HH:mm:ss:SSS z}] %t %5p %c{2}: %m%n

1.3.2 接口

接口类DemoService:

package com.zyz;

public interface DemoService {

    String sayHello(String name);

}

1.3.3 服务消费者

@Reference引入服务，加入版本表示具体引入哪一个实现类的实例

启动类:

package com.zyz;

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(version = "default")
    private DemoService demoService;

    public static void main(String[] args) {
        ConfigurableApplicationContext context = SpringApplication.run(DubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);
        System.out.println((demoService.sayHello("zyz")));
    }

}

配置文件application.yml:

spring:
  application:
    name: dubbo-consumer-demo

server:
  port: 8082

dubbo:
  registry:
    address: zookeeper://127.0.0.1:2181

Zookeeper的配置参考

先启动Zookeeper，再启动provider的启动类，最后启动consumer的启动类

看到输出:

dubbo：20880, Hello, zyz

大功告成。

2. 具体应用

2.1 负载均衡

Dubbo提供了四种负载均衡策略，也可以自定义。

Random LoadBalance:随机，按权重设置随机概率。
RoundRobin LoadBalance:轮询，按公约后的权重设置轮询比率。
LeastActive LoadBalance:最少活跃调用数，相同活跃数的随机，活跃数指调用前后计数差。
ConsistentHash LoadBalance:一致性 Hash，相同参数的请求总是发到同一提供者。

2.2 具体代码

修改provider的配置文件为:

# Spring boot application
spring.application.name=dubbo-provider-demo
server.port=8081

# Base packages to scan Dubbo Component: @org.apache.dubbo.config.annotation.Service
dubbo.scan.base-packages=com.zyz.provider.service
dubbo.application.name=${spring.application.name}


## Dubbo Registry
dubbo.registry.address=zookeeper://127.0.0.1:2181

# Dubbo Protocol
#dubbo.protocol.name=dubbo
#dubbo.protocol.port=20880

dubbo.protocols.p1.id=dubbo1
dubbo.protocols.p1.name=dubbo
dubbo.protocols.p1.port=20881
dubbo.protocols.p1.host=0.0.0.0

dubbo.protocols.p2.id=dubbo2
dubbo.protocols.p2.name=dubbo
dubbo.protocols.p2.port=20882
dubbo.protocols.p2.host=0.0.0.0

dubbo.protocols.p3.id=dubbo3
dubbo.protocols.p3.name=dubbo
dubbo.protocols.p3.port=20883
dubbo.protocols.p3.host=0.0.0.0

服务消费端新增类:

package com.zyz.consumer;

import com.zyz.DemoService;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

@EnableAutoConfiguration
public class LoadBalanceDubboConsumerDemo {

    @Reference(version = "default", loadbalance = "roundrobin")
    private DemoService demoService;

    public static void main(String[] args) {
        ConfigurableApplicationContext context = SpringApplication.run(LoadBalanceDubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);

        roundRobin(demoService);

    }

    /**
     * 负载均衡(轮询)
     * @param demoService
     */
    public static void roundRobin(DemoService demoService){

        for (int i = 0; i < 1000; i++) {
            System.out.println((demoService.sayHello("zyz")));
            try {
                Thread.sleep(1 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

}

最终输出为:

dubbo：20882, Hello, zyz
dubbo：20883, Hello, zyz
dubbo：20881, Hello, zyz
dubbo：20882, Hello, zyz
dubbo：20883, Hello, zyz
dubbo：20881, Hello, zyz

2.2 服务超时

2.2.1 说明

在服务提供者和服务消费者上都可以配置服务超时时间，这两者是不⼀样的。

消费者调⽤⼀个服务，分为三步:

消费者发送请求（⽹络传输）
服务端执⾏服务
服务端返回响应（⽹络传输）

如果在服务端和消费端只在其中⼀⽅配置了timeout，那么没有歧义，表示消费端调⽤服务的超时时间，消费端如果超过时间还没有收到响应结果，则消费端会抛超时异常，但，服务端不会抛异常，服务端在执⾏服务后，会检查执⾏该服务的时间，如果超过timeout，则会打印⼀个超时⽇志。服务会正常的执⾏完。

如果在服务端和消费端各配了⼀个timeout，那就⽐较复杂了，假设

服务执⾏为5s
消费端timeout=3s
服务端timeout=6s

那么消费端调⽤服务时，消费端会收到超时异常（因为消费端超时了），服务端⼀切正常（服务端没有超时）。

2.2.2 具体代码

服务提供方新增两个类:

package com.zyz.provider.service;


import com.zyz.DemoService;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.annotation.Service;
import org.apache.dubbo.rpc.RpcContext;

import java.util.concurrent.TimeUnit;


@Service(version = "timeout", timeout = 6000)
public class TimeoutDemoService implements DemoService {

    @Override
    public String sayHello(String name) {
        System.out.println("执行了timeout服务" + name);

        // 服务执行5秒
        // 服务超时时间为3秒，但是执行了5秒，服务端会把任务执行完的
        // 服务的超时时间，是指如果服务执行时间超过了指定的超时时间则会抛一个warn
        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("执行结束" + name);

        URL url = RpcContext.getContext().getUrl();
        return String.format("%s：%s, Hello, %s", url.getProtocol(), url.getPort(), name);  // 正常访问
    }

}

package com.zyz;

import org.apache.dubbo.config.spring.ServiceBean;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.stereotype.Component;

@Component
public class UpdateBeanPostProcessors implements BeanPostProcessor {

    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {

        if (beanName.contains("ServiceBean")) {
            //这里设置id，否则会造成同一个Service有多个group时，只能注入第一个service
            ServiceBean serviceBean = (ServiceBean) bean;
            serviceBean.setId(beanName);
        }
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }
}

消费方新增一个启动类:

package com.zyz.consumer;

import com.zyz.DemoService;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

import java.io.IOException;

@EnableAutoConfiguration
public class TimeoutDubboConsumerDemo {


    @Reference(version = "timeout", timeout = 3000)
    private DemoService demoService;

    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext context = SpringApplication.run(TimeoutDubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);

        // 服务调用超时时间为1秒，默认为3秒
        // 如果这1秒内没有收到服务结果，则会报错
        System.out.println((demoService.sayHello("周瑜"))); //xxservestub


    }

}

2.3 集群容错

集群容错表示：服务消费者在调⽤某个服务时，这个服务有多个服务提供者，在经过负载均衡后选出其中⼀个服务提供者之后进⾏调⽤，但调⽤报错后，Dubbo所采取的后续处理策略。

2.3.1 设置

Failover Cluster:默认配置，重试两次，一共请求三次
Failfast Cluster:快速失败，只发起一次调用，失败立即报错。通常用于非幂等性的写操作，比如新增记录
Failsafe Cluster:失败安全，出现异常时，直接忽略。通常用于写入审计日志等操作
Failback Cluster:失败自动恢复，后台记录失败请求，定时重发。通常用于消息通知操作
Forking Cluster:并行调用多个服务器，只要一个成功即返回。通常用于实时性要求较高的读操作，但需要浪费更多服务资源。可通过 forks="2" 来设置最大并行数。
Broadcast Cluster:广播调用所有提供者，逐个调用，任意一台报错则报错。通常用于通知所有提供者更新缓存或日志等本地资源信息。

2.3.2 具体代码

cousumer新增一个类:

package com.zyz.consumer;

import com.zyz.DemoService;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

import java.io.IOException;

@EnableAutoConfiguration
public class ClusterDubboConsumerDemo {


    @Reference(version = "cluster", timeout = 1000, cluster = "failfast")
    private DemoService demoService;

    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext context = SpringApplication.run(ClusterDubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);

        System.out.println((demoService.sayHello("zyz")));

    }

}

provider新增一个实现类:

package com.zyz.provider.service;

import com.zyz.DemoService;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.annotation.Service;
import org.apache.dubbo.rpc.RpcContext;

import java.util.concurrent.TimeUnit;

@Service(version = "cluster", timeout = 3000)
public class ClusterDemoService implements DemoService {

    @Override
    public String sayHello(String name) {
        System.out.println("执行了cluster服务" + name);

        try {
            TimeUnit.SECONDS.sleep(5);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        System.out.println("执行结束" + name);
        URL url = RpcContext.getContext().getUrl();
        return String.format("%s：%s, Hello, %s", url.getProtocol(), url.getPort(), name);  // 正常访问
    }
}

2.4 服务降级

服务降级表示：服务消费者在调⽤某个服务提供者时，如果该服务提供者报错了，所采取的措施。

集群容错和服务降级的区别在于：

集群容错是整个集群范围内的容错
服务降级是单个服务提供者的⾃身容错

2.4.1 具体代码

consumer新增一个类:

package com.zyz.consumer;

import com.zyz.DemoService;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

import java.io.IOException;

@EnableAutoConfiguration
public class MockDubboConsumerDemo {


    @Reference(version = "timeout", timeout = 1000, mock = "fail: return 123")
    private DemoService demoService;

    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext context = SpringApplication.run(MockDubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);

        System.out.println((demoService.sayHello("zyz")));
    }

}

2.5 本地存根

本地存根就是⼀段逻辑，这段逻辑是在服务消费端执⾏的，这段逻辑⼀般都是由服务提供者提供，服务提供者可以利⽤这种机制在服务消费者远程调⽤服务提供者之前或之后再做⼀些其他事情，⽐如结果缓存，请求参数验证等等。(AOP功能)

2.5.1 具体代码

interface新增一个类:

package com.zyz;

public class DemoServiceStub implements DemoService {

    private final DemoService demoService;

    // 构造函数传入真正的远程代理对象
    public DemoServiceStub(DemoService demoService){
        this.demoService = demoService;
    }

    @Override
    public String sayHello(String name) {
        // 此代码在客户端执行, 你可以在客户端做ThreadLocal本地缓存，或预先验证参数是否合法，等等
        try {
            System.out.println("校验逻辑");
            return demoService.sayHello(name); // safe  null
        } catch (Exception e) {
            // 你可以容错，可以做任何AOP拦截事项
            return "容错数据";
        }
    }
}

consumer新增一个类:

package com.zyz.consumer;

import com.zyz.DemoService;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

import java.io.IOException;

@EnableAutoConfiguration
public class StubDubboConsumerDemo {


    @Reference(version = "timeout", timeout = 1000, stub = "com.zyz.DemoServiceStub")
    private DemoService demoService;

    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext context = SpringApplication.run(StubDubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);

        System.out.println((demoService.sayHello("周瑜")));


    }

}

2.6 本地伪装

就是Mock功能，Mock其实就是Dubbo中的服务容错的解决⽅案。

2.7 参数回调

⾸先，如果当前服务⽀持参数回调，意思就是：对于某个服务接⼝中的某个⽅法，如果想⽀持消费者在调⽤这个⽅法时能设置回调逻辑，那么该⽅法就需要提供⼀个⼊参⽤来表示回调逻辑。

因为Dubbo协议是基于⻓连接的，所以消费端在两次调⽤同⼀个⽅法时想指定不同的回调逻辑，那么就需要在调⽤时在指定⼀定key进⾏区分。

2.7.1 参数回调

consumer新增两个类

package com.zyz.consumer;


import com.zyz.DemoServiceListener;

public class DemoServiceListenerImpl implements DemoServiceListener {

    @Override
    public void changed(String msg) {
        System.out.println("被回调了："+msg);
    }
}

package com.zyz.consumer;

import com.zyz.DemoService;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

import java.io.IOException;

@EnableAutoConfiguration
public class CallbackDubboConsumerDemo {


    @Reference(version = "callback")
    private DemoService demoService;

    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext context = SpringApplication.run(CallbackDubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);

        // 用来进行callback
        System.out.println(demoService.sayHello("zyz", "d1", new DemoServiceListenerImpl()));
        System.out.println(demoService.sayHello("zyz", "d2", new DemoServiceListenerImpl()));
        System.out.println(demoService.sayHello("zyz", "d3", new DemoServiceListenerImpl()));
    }

}

interface新增一个类，并在DemoService中新增方法:

package com.zyz;

public interface DemoService {

    String sayHello(String name);

    // 添加回调
    default String sayHello(String name, String key, DemoServiceListener listener) {
        return null;
    };

}

package com.zyz;

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

provider新增类:

package com.zyz.provider.service;

import com.zyz.DemoService;
import com.zyz.DemoServiceListener;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.annotation.Argument;
import org.apache.dubbo.config.annotation.Method;
import org.apache.dubbo.config.annotation.Service;
import org.apache.dubbo.rpc.RpcContext;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;


// DemoService的sayHello方法的index=1的参数是回调对象，服务消费者可以调用addListener方法来添加回调对象，服务提供者一旦执行回调对象的方法就会通知给服务消费者

@Service(version = "callback", methods = {@Method(name = "sayHello", arguments = {@Argument(index = 2, callback = true)})}, callbacks = 3)
public class CallBackDemoService implements DemoService {

    private final Map<String, DemoServiceListener> listeners = new ConcurrentHashMap<String, DemoServiceListener>();

    public CallBackDemoService() {
        Thread t = new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    for (Map.Entry<String, DemoServiceListener> entry : listeners.entrySet()) {
                        entry.getValue().changed(getChanged(entry.getKey()));
                    }
                    try {
                        Thread.sleep(5000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        });

        t.start();

    }


    private String getChanged(String key) {
        return "Changed: " + new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
    }

    @Override
    public String sayHello(String name) {
        return null;
    }

    @Override
    public String sayHello(String name, String key, DemoServiceListener callback) {
        System.out.println("执行了回调服务" + name);

        listeners.put(key, callback);
        URL url = RpcContext.getContext().getUrl();
        return String.format("%s：%s, Hello, %s", url.getProtocol(), url.getPort(), name);  // 正常访问
    }

}

2.8 异步调用

从 2.7.0 开始，Dubbo 的所有异步编程接口开始以 CompletableFuture 为基础

基于 NIO 的非阻塞实现并行调用，客户端不需要启动多线程即可完成并行调用多个远程服务，相对多线程开销较小。

2.8.1 具体代码

consumer新增类:

package com.zyz.consumer;

import com.zyz.DemoService;
import org.apache.dubbo.config.annotation.Reference;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

import java.io.IOException;
import java.util.concurrent.CompletableFuture;

@EnableAutoConfiguration
public class AsyncDubboConsumerDemo {

    @Reference(version = "async")
    private DemoService demoService;

    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext context = SpringApplication.run(AsyncDubboConsumerDemo.class);

        DemoService demoService = context.getBean(DemoService.class);

        // 调用直接返回CompletableFuture
        CompletableFuture<String> future = demoService.sayHelloAsync("异步调用");  // 5

        future.whenComplete((v, t) -> {
            if (t != null) {
                t.printStackTrace();
            } else {
                System.out.println("Response: " + v);
            }
        });

        System.out.println("结束了");

    }

}

interface中DemoService新增方法:

package com.zyz;

import java.util.concurrent.CompletableFuture;

public interface DemoService {

    String sayHello(String name);

    // 异步调用方法
    default CompletableFuture<String> sayHelloAsync(String name) {
        return null;
    };

    // 添加回调
    default String sayHello(String name, String key, DemoServiceListener listener) {
        return null;
    };

}

consumer新增一个类:

package com.zyz.provider.service;

import com.zyz.DemoService;
import org.apache.dubbo.common.URL;
import org.apache.dubbo.config.annotation.Service;
import org.apache.dubbo.rpc.RpcContext;

import java.util.concurrent.CompletableFuture;

@Service(version = "async")
public class AsyncDemoService implements DemoService {

    @Override
    public String sayHello(String name) {
        System.out.println("执行了同步服务" + name);
        URL url = RpcContext.getContext().getUrl();
        return String.format("%s：%s, Hello, %s", url.getProtocol(), url.getPort(), name);  // 正常访问
    }

    @Override
    public CompletableFuture<String> sayHelloAsync(String name) {
        System.out.println("执行了异步服务" + name);

        return CompletableFuture.supplyAsync(() -> {
            return sayHello(name);
        });
    }
}

2.9 泛化调用

泛化调⽤可以⽤来做服务测试。

在Dubbo中，如果某个服务想要⽀持泛化调⽤，就可以将该服务的generic属性设置为true，那对于服务消费者来说，就可以不⽤依赖该服务的接⼝，直接利⽤GenericService接⼝来进⾏服务调⽤。

2.9.1 具体代码

consumer新增一个类:

package com.zyz.consumer;

import org.apache.dubbo.config.annotation.Reference;
import org.apache.dubbo.rpc.service.GenericService;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.EnableAutoConfiguration;
import org.springframework.context.ConfigurableApplicationContext;

import java.io.IOException;

@EnableAutoConfiguration
public class GenericDubboConsumerDemo {


    @Reference(id = "demoService", version = "default", interfaceName = "com.zyz.DemoService", generic = true)
    private GenericService genericService;

    public static void main(String[] args) throws IOException {
        ConfigurableApplicationContext context = SpringApplication.run(GenericDubboConsumerDemo.class);

        GenericService genericService = (GenericService) context.getBean("demoService");

        Object result = genericService.$invoke("sayHello", new String[]{"java.lang.String"}, new Object[]{"zyz"});
        System.out.println(result);


    }

}

provider新增一个类:

package com.zyz.provider.service;

import org.apache.dubbo.config.annotation.Service;
import org.apache.dubbo.rpc.service.GenericException;
import org.apache.dubbo.rpc.service.GenericService;

@Service(interfaceName = "com.zyz.DemoService", version = "generic")
public class GenericDemoService implements GenericService {
    @Override
    public Object $invoke(String s, String[] strings, Object[] objects) throws GenericException {
        System.out.println("执行了generic服务");

        return "执行的方法是" + s;
    }
}

3. 相关工具

Zookeeper可视化客户端
 管理台

作者:逍遥白亦

原文链接:https://www.jianshu.com/p/beae47568d1f