java 中间件 基于Netty手写RPC框架进阶版（下）——注册中心及服务的动态扩容

在 基于Netty手写RPC框架进阶版（上）——注解驱动 中，我们搭建好了注解驱动的RPC框架，接下来在本文中我们就来逐步实现注册中心以及动态扩容的功能。

1. 注册中心模块

接下来我们便来引入注册中心，这里使用zookeeper来实现，首先创建netyy-rpc-registry模块。

首先引入zk的相关依赖：

org.apache.curator

curator-framework

5.2.0

org.apache.curator

curator-recipes

5.2.0

org.apache.curator

curator-x-discovery

5.1.0

然后我们定义相关接口，在其中实现注册方法以及发现方法。

public interface IRegistryService {

/**

* 服务注册

*

* @param serviceInfo

*/

void register(ServiceInfo serviceInfo);

/**

* 服务发现

*

* @param serviceName

* @return

*/

ServiceInfo discovery(String serviceName);

}

这里的话我们还需要定义一个实体类封装一些相关的服务信息。

@Data

public class ServiceInfo {

private String serviceName;

private String serviceAddress;

private int servicePort;

}

接下来就是编写实现类了，这里我们可以自行选择使用zookeeper还是eureka，我这里使用zookeeper进行服务的注册。

整个实现类主要通过Curator客户端及ServiceInstance来实现服务的注册以及发现。这里的zookeeper相关代码在之前的文档里面由讲解过，可以自行查看。

public class ZookeeperRegistryService implements IRegistryService {

private static final String REGISTRY_PATH = "/registry";

// curator 中提供的服务注册与发现的封装

private final ServiceDiscovery serviceDiscovery;

// 负载均衡

private ILoadBalance> loadBalance;

public ZookeeperRegistryService(String registryAddress) throws Exception {

// 创建curator客户端

CuratorFramework client = CuratorFrameworkFactory

.newClient(registryAddress, new ExponentialBackoffRetry(1000, 3));

// 启动客户端

client.start();

// 实例化注册发现工具

JsonInstanceSerializer serializer = new JsonInstanceSerializer<>(ServiceInfo.class);

this.serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceInfo.class)

.client(client)

.serializer(serializer)

.basePath(REGISTRY_PATH)

.build();

this.serviceDiscovery.start();

// 实例化负载均衡算法 直接写死，因为这里只有一个

this.loadBalance = new RandomLoadBalance();

}

@Override

public void register(ServiceInfo serviceInfo) throws Exception {

System.out.println("begin registry serviceInfo to zookeeper server");

//将服务端元数据保存到注册中心上

ServiceInstance serviceInstance = ServiceInstance.builder()

.name(serviceInfo.getServiceName())

.address(serviceInfo.getServiceAddress())

.port(serviceInfo.getServicePort())

.payload(serviceInfo)

.build();

//注册

this.serviceDiscovery.registerService(serviceInstance);

}

@Override

public ServiceInfo discovery(String serviceName) throws Exception {

System.out.println("begin discovery serviceInfo from zookeeper server");

Collection> serviceInstances = this.serviceDiscovery.queryForInstances(serviceName);

//动态路由

ServiceInstance serviceInstance = this.loadBalance.select((List>) serviceInstances);

if (serviceInstance == null) {

return null;

}

return serviceInstance.getPayload();

}

}

在发现服务的代码中，我们需要实现动态路由，这里我们可以实现一个随机路由的算法。

首先创建负载均衡接口。

public interface ILoadBalance {

T select(List servers);

}

然后创建一个抽象类作为模板方法：

public abstract class AbstractLoadBalance implements ILoadBalance> {

// 模板方法

@Override

public ServiceInstance select(List> servers) {

if (servers == null || servers.size() == 0) {

return null;

}

if (servers.size() == 1) {

return servers.get(0);

}

// 其他的由具体的子类实现

return doSelect(servers);

}

protected abstract ServiceInstance doSelect(List> servers);

}

不同的路由算法由各个子类单独实现：

public class RandomLoadBalance extends AbstractLoadBalance {

@Override

protected ServiceInstance doSelect(List> servers) {

int len = servers.size();

Random random = new Random();

return servers.get(random.nextInt(len));

}

}

服务注册这里处理完之后，我们来处理一些如何供外部调用。

首先创建类型枚举，让客户端可以选择不同的注册中心。

public enum RegistryType {

ZOOKEEPER((byte) 0),

EUREKA((byte) 1);

private byte code;

RegistryType(byte code) {

this.code = code;

}

public static RegistryType findByCode(int code) {

for (RegistryType value : RegistryType.values()) {

if (value.code == code) {

return value;

}

}

return null;

}

}

创建简易工厂类，根据不同的类型返回不同的注册服务。

public class RegistryFactory {

public static IRegistryService createRegistryService(String address, RegistryType registryType) {

IRegistryService service = null;

try {

switch (registryType) {

case EUREKA:

// todo

break;

case ZOOKEEPER:

service = new ZookeeperRegistryService(address);

break;

default:

service = new ZookeeperRegistryService(address);

break;

}

} catch (Exception e) {

e.printStackTrace();

}

return service;

}

}

2. 服务注册

注册中心模块完成之后，我们就可以到框架模块将相关的服务注册上去了。

首先引入相关依赖：

org.example

netty-rpc-registry

1.0-SNAPSHOT

然后修改SpringRpcProviderBean的相关代码：

首先引入注册中心，然后将相关服务注册上去即可。

public class SpringRpcProviderBean implements InitializingBean, BeanPostProcessor {

private final int serverPort;

private final String serverAddress;

// 服务注册相关

private final IRegistryService registryService; // 服务注册中心

public SpringRpcProviderBean(int serverPort, IRegistryService registryService) throws UnknownHostException {

this.serverPort = serverPort;

// 获取服务器的IP地址

InetAddress address = InetAddress.getLocalHost();

System.out.println("网络地址：" + address.getHostAddress());

this.serverAddress = address.getHostAddress();

this.registryService = registryService;

}

@Override

public void afterPropertiesSet() throws Exception {

System.out.println("begin deploy netty server");

new Thread(() -> {

new NettyServer(this.serverAddress, this.serverPort).startNettyServer();

}).start();

}

// 任何Bean装载到spring 容器的时候都会回调

@Override

public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {

// 只要bean声明了RemoteService注解，则需要把该服务发布到网络上 允许被调用

if (bean.getClass().isAnnotationPresent(RemoteService.class)) {

// 得到所有方法

Method[] methods = bean.getClass().getMethods();

//我们这里需要把具体的发布映射关系保存下来

for (Method method : methods) {

String serviceName = bean.getClass().getInterfaces()[0].getName();

String key = serviceName + "." + method.getName();

BeanMethod beanMethod = new BeanMethod();

beanMethod.setBean(bean);

beanMethod.setMethod(method);

Mediator.beanMethodMap.put(key, beanMethod);

//服务注册相关

//发布到远程服务端

ServiceInfo serviceInfo = new ServiceInfo();

serviceInfo.setServiceAddress(this.serverAddress);

serviceInfo.setServicePort(this.serverPort);

serviceInfo.setServiceName(serviceName);

try {

registryService.register(serviceInfo); // 注册服务

} catch (Exception e) {

System.out.println(serviceName + "registry failed");

e.printStackTrace();

}

}

}

return bean;

}

}

我们需要通过配置文件信息来控制注册中心的地址以及类型：

@Data

@ConfigurationProperties("rpc")

public class RpcServerProperties {

// private String serviceAddress;

private int servicePort;

// 服务注册相关

private String registryAddress;

private byte registryType;

}

然后在注入SpringRpcProviderBean的时候引入对应的注册中心。

@Configuration

@EnableConfigurationProperties(RpcServerProperties.class)

public class RpcProviderAutoConfiguration {

@Bean

public SpringRpcProviderBean springRpcProviderBean(RpcServerProperties rpcServerProperties) throws UnknownHostException {

// 服务注册的时候需要传递注册中心

IRegistryService registryService = RegistryFactory.createRegistryService(rpcServerProperties.getRegistryAddress(), RegistryType.findByCode(rpcServerProperties.getRegistryType()));

return new SpringRpcProviderBean(rpcServerProperties.getServicePort(), registryService);

}

}

最后在生产者端添加对应配置信息并启动项目查看。

server.port=8081

rpc.servicePort=20880

rpc.registryType=0

rpc.registryAddress=127.0.0.1:2181

打开我们的zookeeper客户端查看服务是否注册上去。

3. 消费端处理

既然我们的服务已经注册到注册中心了，那我们的消费端应该如何完善呢？还记得我们消费端Netty连接的地址和端口号都是我们在配置文件中写死的吗？如果是多个服务的话，是不是就不合理了，所以既然我们已经将服务放到注册中心了，那我们能不能直接连接到服务对应的地址上去呢？

没错，我们只需要在服务端根据名字找到对应的服务，然后将Netty客户端的地址设置成服务的地址和端口即可。

接下来逐步修改完善，首先从Netty客户端开始修改。

客户端这里我们不需要从构造方法中传入地址和端口号了，而是在发送消息的时候传入注册中心，然后注册中心根据类名获取到对应的服务，直接使用服务的地址和端口号进行连接。

public class NettyClient {

private final Bootstrap bootstrap;

private final EventLoopGroup eventLoopGroup = new NioEventLoopGroup();

// private String serviceAddress;

//

// private int servicePort;

// public NettyClient(String serviceAddress, int servicePort) {

// System.out.printf("begin init Netty Client,{},{}", serviceAddress, servicePort);

public NettyClient() {

bootstrap = new Bootstrap();

bootstrap.group(eventLoopGroup)

.channel(NioSocketChannel.class)

.handler(new RpcClientInitializer());

// this.serviceAddress = serviceAddress;

// this.servicePort = servicePort;

}

// 发送数据包

// 通过注册服务来连接对应地址

public void sendRequest(RpcProtocol protocol, IRegistryService registryService) throws Exception {

// 服务发现

ServiceInfo discovery = registryService.discovery(protocol.getContent().getClassName());

// 地址都来自注册注册中心 ，不需要通过外部传递了

final ChannelFuture future = bootstrap.connect(discovery.getServiceAddress(), discovery.getServicePort()).sync();

//监听是否连接成功

future.addListener(listener -> {

if (future.isSuccess()) {

// System.out.printf("connect rpc server {} success.", this.serviceAddress);

} else {

// System.out.printf("connect rpc server {} failed. ", this.serviceAddress);

future.cause().printStackTrace();

eventLoopGroup.shutdownGracefully();

}

});

System.out.println("begin transfer data");

future.channel().writeAndFlush(protocol);

}

}

修改完客户端后，接下来就是对动态代理部分的代码进行修改，也是将地址和端口号改为服务的地址和端口。

首先是RpcInvokerProxy类，同样的这里不在传递地址和端口号，改为传递注册中心。

public class RpcInvokerProxy implements InvocationHandler {

// 不需要手动传入了，直接从注册中心获取

private IRegistryService registryService;

// private String host;

//

// private int port;

//

// public RpcInvokerProxy(String host, int port) {

// this.host = host;

// this.port = port;

// }

public RpcInvokerProxy(IRegistryService registryService) {

this.registryService = registryService;

}

@Override

public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {

System.out.println("begin invoke target server");

RpcProtocol reqProtocol = new RpcProtocol<>();

long requestId = RequestHolder.REQUEST_ID.incrementAndGet();

Header header = new Header(RpcConstant.MAGIC, SerialType.JSON_SERIAL.code(),

ReqType.REQUEST.code(), requestId, 0);

reqProtocol.setHeader(header);

RpcRequest request = new RpcRequest();

request.setClassName(method.getDeclaringClass().getName());

request.setMethodName(method.getName());

request.setParamsTypes(method.getParameterTypes());

request.setParams(args);

reqProtocol.setContent(request);

//短链接

// NettyClient nettyClient = new NettyClient(host, port);

// 不需要传递地址及端口了

NettyClient nettyClient = new NettyClient();

// 通过设置DefaultEventLoop进行轮询获取结果

RpcFuture future = new RpcFuture<>(new DefaultPromise(new DefaultEventLoop()));

// 保存请求ID和返回数据的对应关系

RequestHolder.REQUEST_MAP.put(requestId, future);

nettyClient.sendRequest(reqProtocol, registryService);

// 返回异步回调的数据

return future.getPromise().get().getData();

}

}

然后是SpringRpcReferenceBean类，我们需要在这里给RpcInvokerProxy类传递注册中心，这里的话需要我们设置注册中心的地址和端口号。

public class SpringRpcReferenceBean implements FactoryBean