Производительны­е API с Java и gRPC

public class UserServiceImpl extends UserServiceGrpc.UserServiceImplBase {
    private final Map<String, User> userMap = new ConcurrentHashMap<>();
    
    @Override
    public void getUser(UserRequest request, StreamObserver<UserResponse> responseObserver) {
        String userId = request.getUserId();
        User user = userMap.get(userId);
        
        if (user != null) {
            UserResponse response = UserResponse.newBuilder()
                .setUserId(user.getId())
                .setName(user.getName())
                .setEmail(user.getEmail())
                .build();
            
            responseObserver.onNext(response);
            responseObserver.onCompleted();
        } else {
            responseObserver.onError(
                Status.NOT_FOUND.withDescription("Пользователь не найден").asRuntimeException()
            );
        }
    }
    
    @Override
    public void createUser(CreateUserRequest request, StreamObserver<CreateUserResponse> responseObserver) {
        String userId = UUID.randomUUID().toString();
        User user = new User(userId, request.getName(), request.getEmail());
        
        userMap.put(userId, user);
        
        CreateUserResponse response = CreateUserResponse.newBuilder()
            .setUserId(userId)
            .build();
        
        responseObserver.onNext(response);
        responseObserver.onCompleted();
    }
}

public class GrpcServer {
    public static void main(String[] args) throws IOException, InterruptedException {
        Server server = ServerBuilder.forPort(50051)
            .addService(new UserServiceImpl())
            .build();
        
        server.start();
        System.out.println("Сервер запущен на порту " + server.getPort());
        
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            System.out.println("Получен сигнал завершения работы");
            server.shutdown();
            System.out.println("Сервер остановлен");
        }));
        
        server.awaitTermination();
    }
}

public class GrpcClient {
    private final UserServiceGrpc.UserServiceBlockingStub blockingStub;
    private final UserServiceGrpc.UserServiceStub asyncStub;
    
    public GrpcClient(String host, int port) {
        ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port)
            .usePlaintext() // Только для разработки, в продакшене используйте TLS
            .build();
        
        blockingStub = UserServiceGrpc.newBlockingStub(channel);
        asyncStub = UserServiceGrpc.newStub(channel);
    }
    
    public UserResponse getUser(String userId) {
        UserRequest request = UserRequest.newBuilder()
            .setUserId(userId)
            .build();
        
        try {
            return blockingStub.getUser(request);
        } catch (StatusRuntimeException e) {
            System.out.println("Ошибка RPC: " + e.getStatus());
            throw e;
        }
    }
    
    public void createUserAsync(String name, String email, Consumer<CreateUserResponse> callback) {
        CreateUserRequest request = CreateUserRequest.newBuilder()
            .setName(name)
            .setEmail(email)
            .build();
        
        asyncStub.createUser(request, new StreamObserver<CreateUserResponse>() {
            @Override
            public void onNext(CreateUserResponse response) {
                callback.accept(response);
            }
            
            @Override
            public void onError(Throwable throwable) {
                System.out.println("Ошибка при создании пользователя: " + throwable.getMessage());
            }
            
            @Override
            public void onCompleted() {
                System.out.println("Запрос на создание пользователя завершен");
            }
        });
    }
}

@Override
public StreamObserver<ChatMessage> chatStream(StreamObserver<ChatMessage> responseObserver) {
    return new StreamObserver<ChatMessage>() {
        @Override
        public void onNext(ChatMessage message) {
            // При получении сообщения от клиента
            System.out.println("Получено сообщение от пользователя " + message.getUserId());
            
            // Перенаправляем сообщение всем участникам (упрощенная логика)
            ChatMessage response = ChatMessage.newBuilder()
                .setUserId(message.getUserId())
                .setContent(message.getContent())
                .setTimestamp(System.currentTimeMillis())
                .build();
                
            responseObserver.onNext(response);
        }
 
        @Override
        public void onError(Throwable t) {
            System.err.println("Ошибка в потоке чата: " + t.getMessage());
        }
 
        @Override
        public void onCompleted() {
            System.out.println("Клиент завершил поток сообщений");
            responseObserver.onCompleted();
        }
    };
}

public void startChat(String userId) {
    StreamObserver<ChatMessage> requestObserver = asyncStub.chatStream(
        new StreamObserver<ChatMessage>() {
            @Override
            public void onNext(ChatMessage message) {
                System.out.println(message.getUserId() + ": " + message.getContent());
            }
 
            @Override
            public void onError(Throwable t) {
                System.err.println("Ошибка при получении сообщения: " + t.getMessage());
            }
 
            @Override
            public void onCompleted() {
                System.out.println("Сервер завершил отправку сообщений");
            }
        }
    );
 
    // Отправка сообщений через полученный requestObserver
    try {
        for (int i = 0; i < 5; i++) {
            ChatMessage message = ChatMessage.newBuilder()
                .setUserId(userId)
                .setContent("Сообщение #" + i)
                .setTimestamp(System.currentTimeMillis())
                .build();
                
            requestObserver.onNext(message);
            Thread.sleep(1000);
        }
    } catch (InterruptedException e) {
        requestObserver.onError(e);
    }
 
    // Сигнализируем о завершении отправки
    requestObserver.onCompleted();
}

private final ConcurrentMap<String, StreamObserver<ChatMessage>> activeConnections = new ConcurrentHashMap<>();
 
@Override
public StreamObserver<ChatMessage> chatStream(StreamObserver<ChatMessage> responseObserver) {
    String connectionId = UUID.randomUUID().toString();
    activeConnections.put(connectionId, responseObserver);
    
    return new StreamObserver<ChatMessage>() {
        // Реализация методов...
        
        @Override
        public void onCompleted() {
            activeConnections.remove(connectionId);
            responseObserver.onCompleted();
        }
    };
}

Server server = ServerBuilder.forPort(50051)
    .addService(new MyServiceImpl())
    .executor(Executors.newWorkStealingPool())  // Использование ForkJoinPool
    .build();

ManagedChannel channel = ManagedChannelBuilder.forAddress(host, port)
    .usePlaintext()
    .keepAliveTime(30, TimeUnit.SECONDS)
    .keepAliveTimeout(10, TimeUnit.SECONDS)
    .maxInboundMessageSize(16 * 1024 * 1024)  // 16 MB
    .build();

private static final PooledObjectFactory<UserResponse.Builder> BUILDER_POOL =
    new AbstractPooledObjectFactory<UserResponse.Builder>() {
        @Override
        public UserResponse.Builder create() {
            return UserResponse.newBuilder();
        }
        
        @Override
        public void passivateObject(UserResponse.Builder builder) {
            builder.clear();
        }
    };
 
private final ObjectPool<UserResponse.Builder> builderPool = 
    new GenericObjectPool<>(BUILDER_POOL);
 
// Использование пула при создании ответов
UserResponse response = builderPool.borrowObject()
    .setUserId(user.getId())
    .setName(user.getName())
    .setEmail(user.getEmail())
    .build();
builderPool.returnObject(response.toBuilder());

return ReactorGrpc.ReactorUserServiceStub(channel)
    .getUser(request)
    .flatMap(userResponse -> {
        // Обработка ответа
        return Mono.just(transformedResponse);
    })
    .onErrorResume(throwable -> {
        // Обработка ошибок
        return Mono.error(new RuntimeException("Ошибка получения пользователя"));
    });

public class MetricsInterceptor implements ServerInterceptor {
    private final MeterRegistry registry;
    
    @Override
    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
            ServerCall<ReqT, RespT> call,
            Metadata headers,
            ServerCallHandler<ReqT, RespT> next) {
        String method = call.getMethodDescriptor().getFullMethodName();
        Timer.Sample sample = Timer.start(registry);
        
        return new ForwardingServerCallListener.SimpleForwardingServerCallListener<ReqT>(
                next.startCall(new ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
                    @Override
                    public void close(Status status, Metadata trailers) {
                        sample.stop(registry.timer("grpc.server", 
                                "method", method, 
                                "status", status.getCode().name()));
                        super.close(status, trailers);
                    }
                }, headers)) {};
    }
}

public class GrpcApiGateway {
    private final UserServiceGrpc.UserServiceBlockingStub userService;
    private final OrderServiceGrpc.OrderServiceBlockingStub orderService;
 
    public GrpcApiGateway(String userServiceHost, int userServicePort,
                         String orderServiceHost, int orderServicePort) {
        ManagedChannel userChannel = ManagedChannelBuilder
            .forAddress(userServiceHost, userServicePort)
            .usePlaintext()
            .build();
            
        ManagedChannel orderChannel = ManagedChannelBuilder
            .forAddress(orderServiceHost, orderServicePort)
            .usePlaintext()
            .build();
            
        userService = UserServiceGrpc.newBlockingStub(userChannel);
        orderService = OrderServiceGrpc.newBlockingStub(orderChannel);
    }
 
    public OrderDetails getOrderDetailsWithUserInfo(String orderId) {
        // Получаем данные заказа
        OrderResponse order = orderService.getOrder(
            OrderRequest.newBuilder().setOrderId(orderId).build()
        );
        
        // Получаем данные пользователя
        UserResponse user = userService.getUser(
            UserRequest.newBuilder().setUserId(order.getUserId()).build()
        );
        
        // Комбинируем результаты
        return OrderDetails.newBuilder()
            .setOrderInfo(order)
            .setUserInfo(user)
            .build();
    }
}