<?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>grpc-demo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>8</maven.compiler.source>
        <maven.compiler.target>8</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
        <!--grpc版本 -->
        <grpc.version>1.60.2</grpc.version>
    </properties>

    <dependencies>
        <!-- grpc 需要的依賴-->
        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-netty-shaded</artifactId>
            <version>${grpc.version}</version>
        </dependency>

        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-protobuf</artifactId>
            <version>${grpc.version}</version>
        </dependency>

        <dependency>
            <groupId>io.grpc</groupId>
            <artifactId>grpc-stub</artifactId>
            <version>${grpc.version}</version>
        </dependency>

    </dependencies>
    <build>
        <extensions>
            <extension>
                <groupId>kr.motd.maven</groupId>
                <artifactId>os-maven-plugin</artifactId>
                <version>1.5.0.Final</version>
            </extension>
        </extensions>
        <plugins>
            <plugin>
                <groupId>org.xolstice.maven.plugins</groupId>
                <artifactId>protobuf-maven-plugin</artifactId>
                <version>0.5.0</version>
                <configuration>
                    <protocArtifact>com.google.protobuf:protoc:3.3.0:exe:${os.detected.classifier}</protocArtifact>
                    <pluginId>grpc-java</pluginId>
                    <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <!--它將{@code .proto}文件 生成java源代碼 -->
                            <goal>compile</goal>
                            <!--它將{@code .proto}文件 生成grpc java源代碼 -->
                            <goal>compile-custom</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <!-- 解決jar 包沖突、重復(fù)依賴、無效引用 -->
                <artifactId>maven-enforcer-plugin</artifactId>
                <version>1.4.1</version>
                <executions>
                    <execution>
                        <id>enforce</id>
                        <goals>
                            <goal>enforce</goal>
                        </goals>
                        <configuration>
                            <rules>
                                <requireUpperBoundDeps/>
                            </rules>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

// Copyright 2015 The gRPC Authors
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
syntax = "proto3";

option java_multiple_files = true;
option java_package = "io.grpc.examples.helloworld";
option java_outer_classname = "HelloWorldProto";
option objc_class_prefix = "HLW";

package helloworld;

// The greeting service definition.
service Greeter {
  // Sends a greeting
  rpc SayHello (HelloRequest) returns (HelloReply) {}
}

// The request message containing the user's name.
message HelloRequest {
  string name = 1;
}

// The response message containing the greetings
message HelloReply {
  string message = 1;
}

#生成protobuf代碼
mvn protobuf:compile
#生成grpc代碼
mvn protobuf:compile-custom

package io.grpc.examples.helloworld2;

import io.grpc.Grpc;
import io.grpc.InsecureServerCredentials;
import io.grpc.Server;
import io.grpc.examples.helloworld.GreeterGrpc;
import io.grpc.examples.helloworld.HelloReply;
import io.grpc.examples.helloworld.HelloRequest;
import io.grpc.stub.StreamObserver;

import java.io.IOException;
import java.util.Objects;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @Author tanyong
 * @Version HelloWordServer v1.0.0 2024/4/2 11:21 $$
 */
public class HelloWordServer {
    private static int port = 50052;

    /**
     * grpc服務(wù)實例
     */
    private Server server;

    /**
     * 啟動 grpc服務(wù)
     *
     * @throws IOException
     */
    public void start() throws IOException {

        server = Grpc.newServerBuilderForPort(port, InsecureServerCredentials.create())
                // 添加業(yè)務(wù)處理類
                .addService(new GreeterImpl(port))
                .build()
                .start();
        // 注冊了一個JVM關(guān)閉鉤子（Shutdown Hook），當(dāng)Java虛擬機(jī)（JVM）即將關(guān)閉時（無論是正常退出還是非正常退出，如接收到操作系統(tǒng)中斷信號）當(dāng)JVM關(guān)閉時，所有已注冊的關(guān)閉鉤子都將被依次調(diào)用
        Runtime.getRuntime().addShutdownHook(new Thread() {
            @Override
            public void run() {
                // Use stderr here since the logger may have been reset by its JVM shutdown hook.
                System.err.println("*** shutting down gRPC server since JVM is shutting down");
                try {
                    // 優(yōu)雅地停止gRPC服務(wù)器實例
                    HelloWordServer.this.stop();
                } catch (InterruptedException e) {
                    e.printStackTrace(System.err);
                }
                System.err.println("*** server shut down");
            }
        });
    }

    /**
     * 停止grpc 實例
     *
     * @throws InterruptedException
     */
    public void stop() throws InterruptedException {
        if (Objects.nonNull(server)) {
            // 發(fā)起服務(wù)器的關(guān)閉流程，不再接受新的連接和請求，但允許現(xiàn)有連接繼續(xù)完成請求處理
            server.shutdown()
                    // 給予服務(wù)器最長30秒的時間去完成所有待處理的工作，超過這個時間限制，程序?qū)⒗^續(xù)執(zhí)行后續(xù)邏輯，即使服務(wù)器還有任務(wù)未完成
                    // 這樣設(shè)計有助于在應(yīng)用退出時確保資源得到釋放，同時也能防止因某些原因?qū)е碌拈L時間無法關(guān)閉的問題。
                    .awaitTermination(30, TimeUnit.SECONDS);
        }

    }

    /**
     * 確保主線程或者其他調(diào)用者線程會在服務(wù)器完全關(guān)閉之前保持等待狀態(tài)。
     * 在主線程上等待終止，因為grpc庫使用守護(hù)線程
     *
     * @throws InterruptedException
     */
    public void blockUntilShutdown() throws InterruptedException {
        if (Objects.nonNull(server)) {
            server.awaitTermination();
        }
    }

    /**
     * 業(yè)務(wù)處理類
     */
    public static class GreeterImpl extends GreeterGrpc
            .GreeterImplBase {

        private final int port;

        public GreeterImpl(int port) {
            this.port = port;
        }

        /**
         * @param request
         * @param responseObserver 這是gRPC提供的響應(yīng)觀察者對象，用于向客戶端發(fā)送響應(yīng)。服務(wù)端通過調(diào)用其方法將響應(yīng)數(shù)據(jù)發(fā)送給客戶端。
         */
        @Override
        public void sayHello(HelloRequest request, StreamObserver<HelloReply> responseObserver) {
            HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + request.getName() + this.port).build();
            try {
                Thread.sleep(500 + new Random().nextInt(1000));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            // 向客戶端發(fā)送響應(yīng)數(shù)據(jù)，即將創(chuàng)建好的 reply 對象推送給客戶端。
            responseObserver.onNext(reply);
            // 表示響應(yīng)已經(jīng)結(jié)束，沒有更多的數(shù)據(jù)要發(fā)送給客戶端。
            responseObserver.onCompleted();
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        final HelloWordServer helloWordServer = new HelloWordServer();
        helloWordServer.start();
        helloWordServer.blockUntilShutdown();
    }
}

public interface HelloWordConstants {
    String SCHEME = "example";
    String SERVICE_NAME = "lb.example.grpc.io";
}

package io.grpc.examples.helloworld2;

import io.grpc.EquivalentAddressGroup;
import io.grpc.NameResolver;

import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Author tanyong
 * @Version LoadBalanceNameResolver v1.0.0 2024/4/2 16:59 $$
 */
public class LoadBalanceNameResolver extends NameResolver {

    private Listener2 listener;

    private List<InetSocketAddress> socketAddressList;
    private String serviceName;

    private final Map<String, List<InetSocketAddress>> addrStore = new HashMap<>();

    public LoadBalanceNameResolver(List<InetSocketAddress> socketAddressList, String serviceName) {
        this.socketAddressList = socketAddressList;
        this.serviceName = serviceName;
        addrStore.put(serviceName, socketAddressList);

    }

    @Override
    public void start(Listener2 listener) {

        this.listener = listener;
        this.resolve();
    }

    @Override
    public void refresh() {
        this.resolve();
    }

    /**
     * 返回用于對服務(wù)器連接進(jìn)行身份驗證的權(quán)限。它必須來自受信任的來源，因為如果權(quán)限被篡改，rpc可能會被發(fā)送給攻擊者，這可能會泄露敏感的用戶數(shù)據(jù)。
     * 實現(xiàn)必須在不阻塞的情況下生成它，通常是在線生成，并且必須保持它不變。從同一工廠使用相同參數(shù)創(chuàng)建的namesolvers必須返回相同的權(quán)限。
     * 自:
     *
     * @return
     */
    @Override
    public String getServiceAuthority() {
        return "ok";
    }

    @Override
    public void shutdown() {

    }

    /**
     * 解析服務(wù)器地址
     */
    private void resolve() {
        List<InetSocketAddress> addresses = addrStore.get(serviceName);
        List<EquivalentAddressGroup> equivalentAddressGroup = addresses.stream().map(this::toSocketAddress)
                .map(Arrays::asList)
                .map(this::addrToEquivalentAddressGroup)
                .collect(Collectors.toList());
        ResolutionResult resolutionResult = ResolutionResult.newBuilder()
                .setAddresses(equivalentAddressGroup)
                .build();

        // 處理已解析地址和屬性的更新
        this.listener.onResult(resolutionResult);

    }

    private SocketAddress toSocketAddress(InetSocketAddress address) {
        return new InetSocketAddress(address.getHostName(), address.getPort());
    }

    private EquivalentAddressGroup addrToEquivalentAddressGroup(List<SocketAddress> addrList) {
        return new EquivalentAddressGroup(addrList);
    }
}

package io.grpc.examples.helloworld2;

import io.grpc.NameResolver;
import io.grpc.NameResolverProvider;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.List;

/**
 * @Author tanyong
 * @Version LoadBalanceNameResolverProvider v1.0.0 2024/4/3 9:56 $$
 */
public class LoadBalanceNameResolverProvider extends NameResolverProvider {

    private final List<InetSocketAddress> socketAddressList;

    public LoadBalanceNameResolverProvider(List<InetSocketAddress> socketAddressList) {
        this.socketAddressList = socketAddressList;
    }


    @Override
    protected boolean isAvailable() {
        return true;
    }

    @Override
    protected int priority() {
        return 5;
    }

    @Override
    public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
        return new LoadBalanceNameResolver(socketAddressList, HelloWordConstants.SERVICE_NAME);
    }

    @Override
    public String getDefaultScheme() {
        return HelloWordConstants.SCHEME;
    }
}

package io.grpc.examples.helloworld2;

import io.grpc.*;
import io.grpc.examples.helloworld.GreeterGrpc;
import io.grpc.examples.helloworld.HelloReply;
import io.grpc.examples.helloworld.HelloRequest;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @Author tanyong
 * @Version HellowordClient v1.0.0 2024/4/2 16:33 $$
 */
public class HelloWordClient {

    /**
     * 客戶端對服務(wù)Greeter進(jìn)行同步rpc調(diào)用 blockingStub
     */
    private final GreeterGrpc.GreeterBlockingStub blockingStub;


    public HelloWordClient(Channel channel) {
        // 創(chuàng)建一個新的阻塞式Stub，支持對服務(wù)的一元和流輸出調(diào)用
        blockingStub = GreeterGrpc.newBlockingStub(channel);
    }

    public void greet(String name) {
        HelloRequest request = HelloRequest.newBuilder().setName(name).build();

        HelloReply response;
        try {
            // 設(shè)置此次RPC調(diào)用的響應(yīng)超時時間為1秒
            response = blockingStub.withDeadlineAfter(1, TimeUnit.SECONDS).sayHello(request);
            System.out.println(response.getMessage());
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        // 服務(wù)段連接地址
        List<InetSocketAddress> addressList = Arrays.asList(new InetSocketAddress("127.0.0.1", 50052),
                new InetSocketAddress("127.0.0.1", 50053));

        // 注冊 NameResolverProvider
        NameResolverRegistry.getDefaultRegistry().register(new LoadBalanceNameResolverProvider(addressList));
        // 符合namesolver的有效URI
        String target = String.format("%s:///%s", HelloWordConstants.SCHEME, HelloWordConstants.SERVICE_NAME);


        // 創(chuàng)建channel
        ManagedChannel channel = ManagedChannelBuilder.forTarget(target)
                .defaultLoadBalancingPolicy("round_robin")
                // 使用明文連接到服務(wù)器。默認(rèn)情況下，將使用安全連接機(jī)制，如TLS。
                // 應(yīng)僅用于測試或API的使用或交換的數(shù)據(jù)不敏感的API。
                .usePlaintext()
                .disableRetry()
                .build();

        try {
            HelloWordClient client = new HelloWordClient(channel);
            long current = System.currentTimeMillis();
            for (int i = 0; i < 10; i++) {
                Thread.sleep(5000);
                client.greet("測試");
            }
            System.out.println(System.currentTimeMillis() - current);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            // 關(guān)閉channel
            channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
        }

    }
}