private final UserService userService;

public FgeduServiceImpl(UserService userService) {
this.userService = userService;
}

// 简单RPC
@Override
public void sayHello(HelloRequest request, StreamObserver responseObserver) {
String message = “Hello, ” + request.getName() + “! Welcome to gRPC.”;
HelloReply reply = HelloReply.newBuilder()
.setMessage(message)
.build();
responseObserver.onNext(reply);
responseObserver.onCompleted();
}

// 服务端流式RPC
@Override
public void listUsers(UserQuery request, StreamObserver responseObserver) {
List users = userService.findByQuery(request);
for (User user : users) {
responseObserver.onNext(user);
}
responseObserver.onCompleted();
}

// 客户端流式RPC
@Override
public StreamObserver uploadUsers(StreamObserver responseObserver) {
return new StreamObserver() {
private int successCount = 0;
private int failCount = 0;

@Override
public void onNext(User user) {
try {
userService.save(user);
successCount++;
} catch (Exception e) {
failCount++;
}
}

@Override
public void onError(Throwable t) {
responseObserver.onError(t);
}

@Override
public void onCompleted() {
UploadResult result = UploadResult.newBuilder()
.setSuccessCount(successCount)
.setFailCount(failCount)
.build();
responseObserver.onNext(result);
responseObserver.onCompleted();
}
};
}

// 双向流式RPC
@Override
public StreamObserver chat(StreamObserver responseObserver) {
return new StreamObserver() {
@Override
public void onNext(ChatMessage message) {
// 处理消息并回复
ChatMessage reply = ChatMessage.newBuilder()
.setFrom(“Server”)
.setTo(message.getFrom())
.setContent(“Received: ” + message.getContent())
.setTimestamp(System.currentTimeMillis())
.build();
responseObserver.onNext(reply);
}

@Override
public void onError(Throwable t) {
responseObserver.onError(t);
}

@Override
public void onCompleted() {
responseObserver.onCompleted();
}
};
}
}

private final ManagedChannel channel;
private final FgeduServiceGrpc.FgeduServiceBlockingStub blockingStub;
private final FgeduServiceGrpc.FgeduServiceStub asyncStub;

public FgeduClient(String host, int port) {
this.channel = ManagedChannelBuilder.forAddress(host, port)
.usePlaintext()
.build();
this.blockingStub = FgeduServiceGrpc.newBlockingStub(channel);
this.asyncStub = FgeduServiceGrpc.newStub(channel);
}

// 简单RPC调用
public String sayHello(String name) {
HelloRequest request = HelloRequest.newBuilder()
.setName(name)
.build();
HelloReply reply = blockingStub.sayHello(request);
return reply.getMessage();
}

// 服务端流式调用
public void listUsers(UserQuery query) {
Iterator users = blockingStub.listUsers(query);
while (users.hasNext()) {
User user = users.next();
System.out.println(“User: ” + user.getName());
}
}

// 客户端流式调用
public void uploadUsers(List users) throws Exception {
CountDownLatch latch = new CountDownLatch(1);

StreamObserver requestObserver = asyncStub.uploadUsers(
new StreamObserver() {
@Override
public void onNext(UploadResult result) {
System.out.println(“Success: ” + result.getSuccessCount());
}

@Override
public void onError(Throwable t) {
latch.countDown();
}

@Override
public void onCompleted() {
latch.countDown();
}
});

for (User user : users) {
requestObserver.onNext(user);
}
requestObserver.onCompleted();
latch.await(1, TimeUnit.MINUTES);
}

public void shutdown() throws InterruptedException {
channel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
}
}

@Override
public ServerCall.Listener interceptCall(
ServerCall call,
Metadata headers,
ServerCallHandler next) {

// 获取请求头
String token = headers.get(Metadata.Key.of(“token”, Metadata.ASCII_STRING_MARSHALLER));

// 日志记录
System.out.println(“Received call: ” + call.getMethodDescriptor().getFullMethodName());

// 验证token
if (token == null || !validateToken(token)) {
call.close(Status.UNAUTHENTICATED.withDescription(“Invalid token”), headers);
return new ServerCall.Listener() {};
}

return next.startCall(call, headers);
}

private boolean validateToken(String token) {
return “fgedu-token”.equals(token);
}
}

# 客户端拦截器
public class ClientInterceptorImpl implements ClientInterceptor {

@Override
public ClientCall interceptCall(
MethodDescriptor method,
CallOptions callOptions,
Channel next) {

return new ForwardingClientCall.SimpleForwardingClientCall(
next.newCall(method, callOptions)) {

@Override
public void start(Listener responseListener, Metadata headers) {
// 添加请求头
headers.put(Metadata.Key.of(“token”, Metadata.ASCII_STRING_MARSHALLER), “fgedu-token”);
super.start(responseListener, headers);
}
};
}
}

# 可用的负载均衡策略：
# – pick_first：选择第一个可用地址
# – round_robin：轮询
# – weighted_round_robin：加权轮询（需要配置权重）

# 使用NameResolver解析服务地址
public class FgeduNameResolver extends NameResolver {

@Override
public void start(Listener listener) {
List servers = new ArrayList<>();
servers.add(new EquivalentAddressGroup(
new InetSocketAddress(“192.168.1.51”, 50051)));
servers.add(new EquivalentAddressGroup(
new InetSocketAddress(“192.168.1.52”, 50051)));
servers.add(new EquivalentAddressGroup(
new InetSocketAddress(“192.168.1.53”, 50051)));

listener.onAddresses(servers, Attributes.EMPTY);
}

@Override
public void refresh() {
// 刷新服务地址
}

@Override
public void shutdown() {
// 关闭
}
}

# 注册NameResolver
NameResolverRegistry.getDefaultRegistry().register(
new NameResolverProvider() {
@Override
public NameResolver newNameResolver(URI targetUri, NameResolver.Args args) {
return new FgeduNameResolver();
}

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

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

@Override
public String getDefaultScheme() {
return “fgedu”;
}
});

private final Map statusMap =
new ConcurrentHashMap<>();

public HealthServiceImpl() {
statusMap.put(“fgedu.FgeduService”, HealthCheckResponse.ServingStatus.SERVING);
}

@Override
public void check(HealthCheckRequest request,
StreamObserver responseObserver) {

String service = request.getService();
HealthCheckResponse.ServingStatus status = statusMap.get(service);

if (status == null) {
responseObserver.onError(Status.NOT_FOUND
.withDescription(“Service not found: ” + service)
.asRuntimeException());
return;
}

HealthCheckResponse response = HealthCheckResponse.newBuilder()
.setStatus(status)
.build();
responseObserver.onNext(response);
responseObserver.onCompleted();
}

@Override
public void watch(HealthCheckRequest request,
StreamObserver responseObserver) {
// 实现健康检查监听
}

public void setStatus(String service, HealthCheckResponse.ServingStatus status) {
statusMap.put(service, status);
}
}

# 使用grpcurl检查健康状态
$ grpcurl -plaintext 192.168.1.51:50051 grpc.health.v1.Health/Check

# 输出示例：
{
“status”: “SERVING”
}