开源RPC框架如何选型

- Python: 移除Python 3.9支持，最低要求Python 3.10+
- Python: 异步栈支持可观测性(Observability)
- Python: 移除GIL优化，提升并发性能
- Ruby: 移除Ruby 3.1支持，最低要求Ruby 3.2+
- Core: 修复ARM弱内存模型上的竞态条件
- Core: EventEngine Windows平台稳定性增强
- Objective-C: 新增receiveNextMessage API

// user.proto - Protocol Buffers 3语法
syntax = "proto3";

package userservice;

option go_package = "github.com/example/userservice";
option java_multiple_files = true;
option java_package = "com.example.userservice";

// 用户服务定义
service UserService {
  // 一元RPC：获取用户信息
  rpc GetUser(GetUserRequest) returns (User);
  
  // 服务端流式RPC：批量获取用户
  rpc ListUsers(ListUsersRequest) returns (stream User);
  
  // 客户端流式RPC：批量创建用户
  rpc CreateUsers(stream CreateUserRequest) returns (CreateUsersResponse);
  
  // 双向流式RPC：实时用户状态同步
  rpc StreamUserStatus(stream UserStatusRequest) returns (stream UserStatusResponse);
}

message GetUserRequest {
  int64 user_id = 1;
}

message User {
  int64 id = 1;
  string name = 2;
  string email = 3;
  int32 status = 4;
  int64 created_at = 5;
}

// server.go - gRPC Go服务端实现
package main

import (
    "context"
    "log"
    "net"
    
    "google.golang.org/grpc"
    "google.golang.org/grpc/credentials"
    pb "github.com/example/userservice"
)

type userServiceServer struct {
    pb.UnimplementedUserServiceServer
}

func (s *userServiceServer) GetUser(ctx context.Context, req *pb.GetUserRequest) (*pb.User, error) {
    // 业务逻辑实现
    return &pb.User{
        Id:    req.UserId,
        Name:  "John Doe",
        Email: "john@example.com",
    }, nil
}

func (s *userServiceServer) ListUsers(req *pb.ListUsersRequest, stream pb.UserService_ListUsersServer) error {
    // 服务端流式响应
    for i := 0; i < 10; i++ {
        user := &pb.User{Id: int64(i), Name: "User"}
        if err := stream.Send(user); err != nil {
            return err
        }
    }
    return nil
}

func main() {
    // 启用TLS
    creds, err := credentials.NewServerTLSFromFile("server.crt", "server.key")
    if err != nil {
        log.Fatalf("failed to load credentials: %v", err)
    }
    
    lis, err := net.Listen("tcp", ":50051")
    if err != nil {
        log.Fatalf("failed to listen: %v", err)
    }
    
    s := grpc.NewServer(grpc.Creds(creds))
    pb.RegisterUserServiceServer(s, &userServiceServer{})
    
    log.Println("gRPC server listening on :50051")
    if err := s.Serve(lis); err != nil {
        log.Fatalf("failed to serve: %v", err)
    }
}

// user.thrift
namespace java com.example.thrift
namespace go example.thrift

struct User {
    1: i64 id,
    2: string name,
    3: string email,
    4: i32 status
}

struct GetUserRequest {
    1: i64 userId
}

service UserService {
    User getUser(1: GetUserRequest request),
    list<User> listUsers(1: i32 limit),
    void createUser(1: User user)
}

+----------------+----------------+----------------+
|  Header(16B)   |  Meta Header   |  Data Payload  |
+----------------+----------------+----------------+
|  Magic(2B)     |  Version(1B)   |  Type(1B)      |
|  Reserved(4B)  |  TotalLen(4B)  |  MetaLen(2B)   |
|  DataLen(2B)   |  StreamID(2B)  |                |
+----------------+----------------+----------------+

// dubbo-triple.proto
syntax = "proto3";

package org.apache.dubbo.demo;

option java_multiple_files = true;
option java_package = "org.apache.dubbo.demo";

service DemoService {
    rpc sayHello(HelloRequest) returns (HelloReply);
    rpc sayHelloStream(HelloRequest) returns (stream HelloReply);
}

message HelloRequest {
    string name = 1;
}

message HelloReply {
    string message = 1;
}

// DemoServiceImpl.java - Dubbo 3.3服务实现
package org.apache.dubbo.demo;

import org.apache.dubbo.config.annotation.DubboService;
import org.apache.dubbo.rpc.RpcContext;
import reactor.core.publisher.Flux;

@DubboService
public class DemoServiceImpl implements DemoService {
    
    @Override
    public HelloReply sayHello(HelloRequest request) {
        return HelloReply.newBuilder()
            .setMessage("Hello " + request.getName())
            .build();
    }
    
    // 流式响应（Mutiny Reactive - Dubbo 3.3.6+）
    @Override
    public Flux<HelloReply> sayHelloStream(HelloRequest request) {
        return Flux.range(1, 10)
            .map(i -> HelloReply.newBuilder()
                .setMessage("Hello " + request.getName() + " - " + i)
                .build());
    }
}

// Consumer配置 - Spring Boot集成
@Configuration
public class DubboConsumerConfig {
    
    @DubboReference(
        interfaceClass = DemoService.class,
        version = "1.0.0",
        timeout = 3000,
        retries = 2,
        loadbalance = "roundrobin"
    )
    private DemoService demoService;
    
    // 虚拟线程池配置（Java 21+）
    @Bean
    public ExecutorService virtualThreadExecutor() {
        return Executors.newVirtualThreadPerTaskExecutor();
    }
}

// UserClient.java - OpenFeign接口定义
@FeignClient(
    name = "user-service",
    url = "${user-service.url}",
    configuration = FeignConfig.class,
    fallbackFactory = UserClientFallback.class
)
public interface UserClient {
    
    @GetMapping("/api/users/{id}")
    UserDTO getUser(@PathVariable("id") Long id);
    
    @PostMapping("/api/users")
    UserDTO createUser(@RequestBody UserCreateRequest request);
    
    @GetMapping("/api/users")
    PageResult<UserDTO> listUsers(@RequestParam("page") int page, 
                                   @RequestParam("size") int size);
}

// 降级处理
@Component
public class UserClientFallback implements FallbackFactory<UserClient> {
    @Override
    public UserClient create(Throwable cause) {
        return new UserClient() {
            @Override
            public UserDTO getUser(Long id) {
                return UserDTO.builder()
                    .id(id)
                    .name("默认用户")
                    .fallback(true)
                    .build();
            }
            // ... 其他方法
        };
    }
}

// ReactiveUserClient.java - WebClient响应式调用
@Service
public class ReactiveUserClient {
    
    private final WebClient webClient;
    
    public ReactiveUserClient(@Value("${user-service.url}") String baseUrl) {
        this.webClient = WebClient.builder()
            .baseUrl(baseUrl)
            .defaultHeader(HttpHeaders.CONTENT_TYPE, MediaType.APPLICATION_JSON_VALUE)
            .filter(new RetryFilter())
            .build();
    }
    
    public Mono<UserDTO> getUser(Long id) {
        return webClient.get()
            .uri("/api/users/{id}", id)
            .retrieve()
            .bodyToMono(UserDTO.class)
            .timeout(Duration.ofSeconds(3))
            .onErrorResume(WebClientResponseException.class, 
                e -> Mono.just(UserDTO.fallback(id)));
    }
    
    public Flux<UserDTO> listUsers(int page, int size) {
        return webClient.get()
            .uri(uriBuilder -> uriBuilder
                .path("/api/users")
                .queryParam("page", page)
                .queryParam("size", size)
                .build())
            .retrieve()
            .bodyToFlux(UserDTO.class);
    }
}

// gRPC服务端集成
@GrpcService
public class UserGrpcService extends UserServiceGrpc.UserServiceImplBase {
    
    @Autowired
    private UserService userService;
    
    @Override
    public void getUser(GetUserRequest request, 
                        StreamObserver<User> responseObserver) {
        User user = userService.findById(request.getUserId());
        responseObserver.onNext(user);
        responseObserver.onCompleted();
    }
}

// gRPC客户端集成
@GrpcClient("user-service")
private UserServiceGrpc.UserServiceBlockingStub userStub;

public User getUser(Long id) {
    GetUserRequest request = GetUserRequest.newBuilder()
        .setUserId(id)
        .build();
    return userStub.getUser(request);
}