关于dubbo:Dubbo的简单使用以及Triple协议的Streaming通信的实现

商业转载请分割作者取得受权，非商业转载请注明出处。
For commercial use, please contact the author for authorization. For non-commercial use, please indicate the source.
协定(License)：署名-非商业性应用-雷同形式共享 4.0 国内 (CC BY-NC-SA 4.0)
作者(Author)：Waste Code
链接(URL)：https://waste-code.tech/archi…
起源(Source)：Waste Code

文章概览

我的项目模块
1. common模块——实现实体类以及申明裸露的api接口
2. provider模块——裸露的api接口的业务实现
3. consumer模块——申请接口的实现，将会待用裸露的api接口
GITHUB: Dubbo的简略应用以及Triple协定的Streaming通信的实现
官网文档: Triple协定
Blog目标: 记录实现过程及呈现的问题

Dubbo的简略应用

在common模块中定义实体类User

在common模块中申明暴露出的接口，实现接口UserService

public interface UserService {

    /**
    * 获取用户信息
    * @param name
    * @return
    */
    User getUserInfo(String name);
}

在provider和consumer模块中引入相干依赖

<dependency>
    <groupId>org.apache.dubbo</groupId>
    <artifactId>dubbo-spring-boot-starter</artifactId>
    <version>3.0.7</version>
</dependency>

<!--   上面这个包必须援用，服务注册到zookeeper中应用，之前没有援用这个包，后果利用起不来         -->
<dependency>
    <groupId>org.apache.dubbo</groupId>
    <artifactId>dubbo-registry-zookeeper</artifactId>
    <version>3.0.7</version>
</dependency>

<dependency>
    <groupId>com.sample</groupId>
    <artifactId>common</artifactId>
    <version>0.0.1-SNAPSHOT</version>
</dependency>

在provider和consumer模块中创立application.yml文件并编写相干配置

server:
    port: 8082 # 这里填写端口号，provider和consumer不同，
spring:
    application:
        name: consumer

dubbo:
    protocol:
        name: dubbo # 抉择通信协议
        port: -1
    registry:
        id: zk-zookeeper
        address: zookeeper://127.0.0.1:2181

在provider和consumer中编写启动类，这里以consumer模块为例，这里要加上EnableDubbo注解

@SpringBootApplication
@EnableDubbo
public class ConsumerApplication {

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

在provider中对UserService进行实现

// 这里留神应用注解@DubboService，时dubbo中的Service注解，次要在对外提供服务的实现类上
@DubboService
public class UserServiceImpl implements UserService {
    @Override
    public User getUserInfo(String name) {
        User user = new User();
        user.setName("dubbo");
        user.setAge(12);
        return user;
    }

}

在consumer中实现申请接口, 援用provider模块暴露出的接口要应用DubboReference注解

@RestController
@RequestMapping("/user")
public class UserController {

    @DubboReference
    private UserService userService;

    @GetMapping("/info")
    public User getUserInfo() {
        return userService.getUserInfo("xxx");
    }


}

编写实现代码后，启动provider和consumer模块，而后通过Postman工具调用接口，发现能够失常应用就实现了

Triple协定的Streaming通信实现

Triple协定的Stream通信次要分为三种：服务端流、客户端流、双向流

利用场景

接口须要发送大量数据，无奈被放到一次申请中，须要分批次发送
流式场景，数据须要依照发送程序解决, 数据自身是没有确定边界的
推送类场景，多个音讯在同一个调用的上下文中被发送和解决

流的语义保障(长处)
提供音讯边界，能够不便的对音讯进行独自解决
严格有序，发送端的程序和接收端的程序是统一的
全双工，发送不须要期待
反对勾销和超时

Streaming流通信实现

服务端流(SERVER_STREAM)申请流程

服务端流(SERVER_STREAM)的Java实现

在provider和consumer模块中增加相干依赖

<dependency>
    <groupId>com.google.protobuf</groupId>
    <artifactId>protobuf-java</artifactId>
</dependency>

批改provider和consumer模块中的相干配置

dubbo: #此处仅截取须要变更的配置，其余配置默认为有原有的就行
    protocol:
        name: tri # 批改dubbo的通信协议，当然triple协定同样反对之前的dubbo的简略应用

在common模块的UserService中申明相干api接口

/**
 * 服务端流
 * @param name
 * @param response
 */
void sayHelloServerStream(String name, StreamObserver<String> response)
    throws InterruptedException;

在provider模块中实现相干性能

//StreamObserver是接管音讯的观察者，
//在onNext办法调用后，consumer模块中的消费者会获取相干的数据，
//当onCompleted办法调用后，consumer模块进行最初的解决后，整个服务端流才会完结
@Override
public void sayHelloServerStream(String name, StreamObserver<String> response)
    throws InterruptedException {

    response.onNext("Hallo, " + name);
    
    // 这里提早10s，次要测试，provider模块接收数据会不会有10s的延时
    Thread.sleep(10 * 1000);

    response.onNext("Hallo, " + name + ", 第二次");

    response.onCompleted();
}

在consumer模块编写申请办法


/**
 * 测试服务端流
 * @param name
 * @return
 * @throws InterruptedException
 */
@GetMapping("/sayHallo/{name}")
public List<String> sayHallo(@PathVariable("name") String name) throws InterruptedException {

    List<String> list = new ArrayList<>();
    userService.sayHelloServerStream(name, new StreamObserver<String>() {
        
        // 每次provider模块调用一次onNext时，该办法会执行一次
        @Override
        public void onNext(String data) {
            System.out.println("onNext:" + data);
            list.add(data);
        }

        @Override
        public void onError(Throwable throwable) {
            System.out.println("报错了");
        }
        // 当provider模块的onCompleted办法调用后，执行该办法
        @Override
        public void onCompleted() {
            System.out.println("完结");
        }
    });
    return list;
}

客户端(CLIENT_STREAM)流申请流程

双向流(BIDIRECTIONAL_STREAM)申请流程

客户端流(CLIENT_STREAM)/双向流(BIDIRECTIONAL_STREAM)的Java实现

客户端流和双向流在Java中的实现形式是同一种
援用pom和批改配置与服务端流雷同

在common模块中申明相干接口

/**
 * 客户端流/双向流, 这里返回的StreamObserver类里的解决切实provider模块中实现，
 * 而参数StreamObserver则是在consumer模块中实现，尽管是consumer调用该办法
 * @param response
 * @return
 */
StreamObserver<String> sayHelloStream(StreamObserver<String> response);

在provider模块中实现相干办法

@Override
public StreamObserver<String> sayHelloStream(StreamObserver<String> response) {
    return new StreamObserver<String>() {
        @Override
        public void onNext(String data) {
            System.out.println("服务端申请参数:" + data);
            response.onNext("Hello, " + data);
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onCompleted() {
            System.out.println("provider敞开");
            response.onCompleted();
        }
    };
}

在consumer模块中实现办法的调用


@PostMapping("/sayHallo")
public List<String> sayHallo(@RequestBody List<String> names) {
    List<String> list = new ArrayList<>();
    StreamObserver<String> request = userService.sayHelloStream(new StreamObserver<String>() {
        @Override
        public void onNext(String data) {
            System.out.println("说了啥？" + data);
            list.add(data);
        }

        @Override
        public void onError(Throwable throwable) {

        }

        @Override
        public void onCompleted() {
            System.out.println("完结了");
        }
    });

    // 下面定义了StreamObserver并调用了办法后，在下边通过onNext办法调用发送申请
    names.forEach(item -> {
        request.onNext(item);
        try {
            Thread.sleep(10 * 1000);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    });
    request.onCompleted();

    return list;
}

关于dubbo:Dubbo的简单使用以及Triple协议的Streaming通信的实现

文章概览

Dubbo的简略应用

Triple协定的Streaming通信实现

利用场景

流的语义保障(长处)

Streaming流通信实现

服务端流(SERVER_STREAM)申请流程

服务端流(SERVER_STREAM)的Java实现

客户端(CLIENT_STREAM)流申请流程

双向流(BIDIRECTIONAL_STREAM)申请流程

客户端流(CLIENT_STREAM)/双向流(BIDIRECTIONAL_STREAM)的Java实现

评论

发表回复取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

关于dubbo:Dubbo的简单使用以及Triple协议的Streaming通信的实现

文章概览

Dubbo的简略应用

Triple协定的Streaming通信实现

利用场景

流的语义保障(长处)

Streaming流通信实现

服务端流(SERVER_STREAM)申请流程

服务端流(SERVER_STREAM)的Java实现

客户端(CLIENT_STREAM)流申请流程

双向流(BIDIRECTIONAL_STREAM)申请流程

客户端流(CLIENT_STREAM)/双向流(BIDIRECTIONAL_STREAM)的Java实现

评论

发表回复 取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

发表回复取消回复