乐趣区

关于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 的简略应用

  1. 在 common 模块中定义实体类 User
  2. 在 common 模块中申明暴露出的接口,实现接口 UserService

    public interface UserService {
    
        /**
        * 获取用户信息
        * @param name
        * @return
        */
        User getUserInfo(String name);
    }
  3. 在 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>
  4. 在 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
  5. 在 provider 和 consumer 中编写启动类,这里以 consumer 模块为例,这里要加上 EnableDubbo 注解

    @SpringBootApplication
    @EnableDubbo
    public class ConsumerApplication {public static void main(String[] args) {SpringApplication.run(ConsumerApplication.class, args);
        }
    }
  6. 在 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;
        }
    
    }
    
  7. 在 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 实现

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

    <dependency>
        <groupId>com.google.protobuf</groupId>
        <artifactId>protobuf-java</artifactId>
    </dependency>
  2. 批改 provider 和 consumer 模块中的相干配置

    dubbo: #此处仅截取须要变更的配置,其余配置默认为有原有的就行
        protocol:
            name: tri # 批改 dubbo 的通信协议,当然 triple 协定同样反对之前的 dubbo 的简略应用
  3. 在 common 模块的 UserService 中申明相干 api 接口

    /**
     * 服务端流
     * @param name
     * @param response
     */
    void sayHelloServerStream(String name, StreamObserver<String> response)
        throws InterruptedException;
  4. 在 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();}
  5. 在 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 实现

  1. 客户端流和双向流在 Java 中的实现形式是同一种
  2. 援用 pom 和批改配置与服务端流雷同
  3. 在 common 模块中申明相干接口

    /**
     * 客户端流 / 双向流, 这里返回的 StreamObserver 类里的解决切实 provider 模块中实现,* 而参数 StreamObserver 则是在 consumer 模块中实现,尽管是 consumer 调用该办法
     * @param response
     * @return
     */
    StreamObserver<String> sayHelloStream(StreamObserver<String> response);
  4. 在 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();}
        };
    }
  5. 在 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;
    }
退出移动版