从零开始手写 dubbo rpc 框架-08-register注册中心

约定不等于承诺〃 2023-05-29 02:47 33阅读 0赞

# register 注册中心 #

## 需求 ##

服务的注册与发现，是一个很常见也很有用的需求。

可以让我们不关心服务端的 ip 信息，只关心服务本身。

# 思路 #

## 实现的方式 ##

SimpleRpcRegister 为 rpc 提供的默认实现方案。

实际可以结合 redis，zk 等常见的成熟框架实现。

其实可以把 register 当做是服务端，此时的 server/client 都是客户端。

实现的策略时类似的。

当然也可以直接使用 zk 等成熟的框架，只是个人觉得这样不利于学习，而且 zk 太重了。

## 流程 ##

*  启动注册中心

首先启动注册中心

*  启动服务端

服务端启动时，将注册信息注册到注册中心。

*  启动客户端

客户端启动的时候，去注册中心读取配置。

# 默认实现 #

## SimpleRpcRegister ##

源码如下，主要是一些注册/取消注册。

订阅和取消订阅。

/** * <p> 简单的 rpc 注册 </p> * * <pre> Created: 2019/10/23 8:59 下午 </pre> * <pre> Project: rpc </pre> * * （1）各种关系的关系服务类 * （2）各种关系之间的通讯类 * （3）domain 层 * * @author houbinbin * @since 0.0.8 */
    public class SimpleRpcRegister implements RpcRegister { 
    
        private static final Log LOG = LogFactory.getLog(DefaultServerRegisterService.class);
    
        /** * 服务端信息管理 * @since 0.0.8 */
        private final ServerRegisterService serverRegisterService;
    
        /** * 客户端信息管理 * @since 0.0.8 */
        private final ClientRegisterService clientRegisterService;
    
        public SimpleRpcRegister(ServerRegisterService serverRegisterService, ClientRegisterService clientRegisterService) { 
            this.serverRegisterService = serverRegisterService;
            this.clientRegisterService = clientRegisterService;
        }
    
        @Override
        public void register(ServiceEntry serviceEntry) { 
            List<ServiceEntry> serviceEntryList = serverRegisterService.register(serviceEntry);
    
            // 通知监听者
            clientRegisterService.notify(serviceEntry.serviceId(), serviceEntryList);
        }
    
        @Override
        public void unRegister(ServiceEntry serviceEntry) { 
            List<ServiceEntry> serviceEntryList = serverRegisterService.unRegister(serviceEntry);
    
            // 通知监听者
            clientRegisterService.notify(serviceEntry.serviceId(), serviceEntryList);
        }
    
        @Override
        public void subscribe(ServiceEntry clientEntry, final Channel channel) { 
            clientRegisterService.subscribe(clientEntry, channel);
        }
    
        @Override
        public void unSubscribe(ServiceEntry clientEntry, Channel channel) { 
            clientRegisterService.unSubscribe(clientEntry, channel);
        }
    
        @Override
        public void lookUp(String seqId, ServiceEntry clientEntry, Channel channel) { 
            final String serviceId = clientEntry.serviceId();
            List<ServiceEntry> serviceEntryList = serverRegisterService.lookUp(serviceId);
    
            // 回写
            // 为了复用原先的相应结果，此处直接使用 rpc response
            RpcResponse rpcResponse = DefaultRpcResponse.newInstance().seqId(seqId)
                    .result(serviceEntryList);
            channel.writeAndFlush(rpcResponse);
        }
    }

## 服务端 ##

服务启动的时候，进行服务注册。

/** * 注冊服務到注册中心 * （1）循环服务列表注册到配置中心列表 * （2）如果 register 为 false，则不进行注册 * （3）后期可以添加延迟暴露，但是感觉意义不大。 * @since 0.0.8 */
        private void registerServiceCenter() { 
            // 注册到配置中心
            // 初期简单点，直接循环调用即可
            // 循环服务信息
            for(ServiceConfig config : this.serviceConfigList) { 
                boolean register = config.register();
                final String serviceId = config.id();
                if(!register) { 
                    LOG.info("[Rpc Server] serviceId: {} register config is false.",
                            serviceId);
                    continue;
                }
    
                for(RpcAddress rpcAddress : registerCenterList) { 
                    ChannelHandler registerHandler = ChannelHandlers.objectCodecHandler(new RpcServerRegisterHandler());
                    LOG.info("[Rpc Server] start register to {}:{}", rpcAddress.address(),
                            rpcAddress.port());
                    ChannelFuture channelFuture = DefaultNettyClient.newInstance(rpcAddress.address(), rpcAddress.port(),registerHandler).call();
    
                    // 直接写入信息
                    RegisterMessage registerMessage = buildRegisterMessage(config);
                    LOG.info("[Rpc Server] register to service center: {}", registerMessage);
                    channelFuture.channel().writeAndFlush(registerMessage);
                }
            }
        }

## 客户端 ##

如果用户没有手动指定 ip 信息，又指定了订阅注册中心。

那么就回去注册中心读取配置，核心代码如下：

/** * 查询服务信息列表 * @return 服务明细列表 * @since 0.0.8 */
        @SuppressWarnings("unchecked")
        private List<ServiceEntry> lookUpServiceEntryList() { 
            //1. 连接到注册中心
            List<ChannelFuture> channelFutureList = connectRegisterCenter();
    
            //2. 选择一个
            // 直接取第一个即可，后续可以使用 load-balance 策略。
            ChannelFuture channelFuture = channelFutureList.get(0);
    
            //3. 发送查询请求
            ServiceEntry serviceEntry = ServiceEntryBuilder.of(serviceId);
            RegisterMessage registerMessage = RegisterMessages.of(MessageTypeConst.CLIENT_LOOK_UP, serviceEntry);
            final String seqId = registerMessage.seqId();
            invokeService.addRequest(seqId, registerCenterTimeOut);
            channelFuture.channel().writeAndFlush(registerMessage);
    
            //4. 等待查询结果
            RpcResponse rpcResponse = invokeService.getResponse(seqId);
            return (List<ServiceEntry>) RpcResponses.getResult(rpcResponse);
        }

# 测试代码 #

## 注册中心 ##

RegisterBs.newInstance().start();

*  日志

[INFO] [2019-11-01 17:36:31.668] [pool-1-thread-1] [c.g.h.r.c.r.n.i.DefaultNettyServer.run] - [Netty Server] start with port: 8527 and channelHandler:
    十一月 01, 2019 5:36:32 下午 io.netty.handler.logging.LoggingHandler channelRegistered
    信息: [id: 0xb7560fe2] REGISTERED
    十一月 01, 2019 5:36:32 下午 io.netty.handler.logging.LoggingHandler bind
    信息: [id: 0xb7560fe2] BIND: 0.0.0.0/0.0.0.0:8527
    十一月 01, 2019 5:36:32 下午 io.netty.handler.logging.LoggingHandler channelActive
    信息: [id: 0xb7560fe2, L:/0:0:0:0:0:0:0:0:8527] ACTIVE
    [INFO] [2019-11-01 17:36:32.655] [pool-1-thread-1] [c.g.h.r.c.r.n.i.DefaultNettyServer.run] - [Netty Server] 启动完成，监听【8527】端口

## 服务端 ##

public static void main(String[] args) { 
        // 启动服务
        DefaultServiceRegistry.getInstance()
                .register(ServiceIdConst.CALC, new CalculatorServiceImpl())
                .registerCenter(ServiceIdConst.REGISTER_CENTER)
                .expose();
    }

*  日志

[INFO] [2019-11-01 17:37:22.116] [main] [c.g.h.r.s.r.i.DefaultServiceRegistry.registerServiceCenter] - [Rpc Server] register to service center: DefaultRegisterMessage{header=DefaultRegisterMessageHeader{type=1}, body=DefaultServiceEntry{serviceId='calc', description='null', ip='172.17.160.1', port=9527, weight=0}}
    [INFO] [2019-11-01 17:37:22.122] [main] [c.g.h.r.s.r.i.DefaultServiceRegistry.expose] - server service register finish.

## 客户端 ##

config.subscribe(true);
    config.registerCenter(ServiceIdConst.REGISTER_CENTER);

*  日志

[INFO] [2019-11-01 17:38:16.279] [main] [c.g.h.r.c.c.r.i.DefaultReferenceConfig.connectRegisterCenter] - [Rpc Client] connect to register 127.0.0.1:8527
    [INFO] [2019-11-01 17:38:16.281] [main] [c.g.h.r.c.r.n.i.DefaultNettyClient.call] - [Netty Client] 开始启动客户端
    [INFO] [2019-11-01 17:38:17.200] [main] [c.g.h.r.c.r.n.i.DefaultNettyClient.call] - [Netty Client] 启动客户端完成，监听地址 127.0.0.1:8527
    [INFO] [2019-11-01 17:38:17.227] [main] [c.g.h.r.c.i.i.DefaultInvokeService.addRequest] - [Client] start add request for seqId: 99835f206f5a45018436c0d6fc0eb71a, timeoutMills: 60000
    [INFO] [2019-11-01 17:38:17.233] [main] [c.g.h.r.c.i.i.DefaultInvokeService.getResponse] - [Client] seq 99835f206f5a45018436c0d6fc0eb71a 对应结果为空，进入等待
    [INFO] [2019-11-01 17:38:17.282] [nioEventLoopGroup-2-1] [c.g.h.r.c.h.RpcClientRegisterHandler.channelRead0] - [Client Register] response is :DefaultRpcResponse{seqId='99835f206f5a45018436c0d6fc0eb71a', error=null, result=[DefaultServiceEntry{serviceId='calc', description='null', ip='172.17.160.1', port=9527, weight=0}]}
    [INFO] [2019-11-01 17:38:17.283] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] 获取结果信息，seqId: 99835f206f5a45018436c0d6fc0eb71a, rpcResponse: DefaultRpcResponse{seqId='99835f206f5a45018436c0d6fc0eb71a', error=null, result=[DefaultServiceEntry{serviceId='calc', description='null', ip='172.17.160.1', port=9527, weight=0}]}
    [INFO] [2019-11-01 17:38:17.283] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:99835f206f5a45018436c0d6fc0eb71a 信息已经放入，通知所有等待方
    [INFO] [2019-11-01 17:38:17.284] [nioEventLoopGroup-2-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:99835f206f5a45018436c0d6fc0eb71a remove from request map
    [INFO] [2019-11-01 17:38:17.284] [main] [c.g.h.r.c.i.i.DefaultInvokeService.getResponse] - [Client] seq 99835f206f5a45018436c0d6fc0eb71a 对应结果已经获取: DefaultRpcResponse{seqId='99835f206f5a45018436c0d6fc0eb71a', error=null, result=[DefaultServiceEntry{serviceId='calc', description='null', ip='172.17.160.1', port=9527, weight=0}]}
    [INFO] [2019-11-01 17:38:17.287] [main] [c.g.h.r.c.c.r.i.DefaultReferenceConfig.getRpcAddresses] - [Client] register center serviceEntries: [DefaultServiceEntry{serviceId='calc', description='null', ip='172.17.160.1', port=9527, weight=0}]
    [INFO] [2019-11-01 17:38:17.288] [main] [c.g.h.r.c.r.n.i.DefaultNettyClient.call] - [Netty Client] 开始启动客户端
    [INFO] [2019-11-01 17:38:17.312] [main] [c.g.h.r.c.r.n.i.DefaultNettyClient.call] - [Netty Client] 启动客户端完成，监听地址 172.17.160.1:9527
    [INFO] [2019-11-01 17:38:17.323] [main] [c.g.h.r.c.p.ReferenceProxy.invoke] - [Client] start call remote with request: DefaultRpcRequest{seqId='ac246a40535e41ddb27768be80e5b381', createTime=1572601097316, serviceId='calc', methodName='sum', paramTypeNames=[com.github.houbb.rpc.server.facade.model.CalculateRequest], paramValues=[CalculateRequest{one=10, two=20}]}
    [INFO] [2019-11-01 17:38:17.323] [main] [c.g.h.r.c.i.i.DefaultInvokeService.addRequest] - [Client] start add request for seqId: ac246a40535e41ddb27768be80e5b381, timeoutMills: 60000
    [INFO] [2019-11-01 17:38:17.341] [main] [c.g.h.r.c.p.ReferenceProxy.invoke] - [Client] start call channel id: 00155dfffe2b601d-000046a8-00000001-279d0e44ded16c5f-efc96a5a
    [INFO] [2019-11-01 17:38:17.349] [main] [c.g.h.r.c.i.i.DefaultInvokeService.getResponse] - [Client] seq ac246a40535e41ddb27768be80e5b381 对应结果为空，进入等待
    [INFO] [2019-11-01 17:38:17.365] [nioEventLoopGroup-4-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] 获取结果信息，seqId: ac246a40535e41ddb27768be80e5b381, rpcResponse: DefaultRpcResponse{seqId='ac246a40535e41ddb27768be80e5b381', error=null, result=CalculateResponse{success=true, sum=30}}
    [INFO] [2019-11-01 17:38:17.366] [nioEventLoopGroup-4-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:ac246a40535e41ddb27768be80e5b381 信息已经放入，通知所有等待方
    [INFO] [2019-11-01 17:38:17.366] [nioEventLoopGroup-4-1] [c.g.h.r.c.i.i.DefaultInvokeService.addResponse] - [Client] seqId:ac246a40535e41ddb27768be80e5b381 remove from request map
    [INFO] [2019-11-01 17:38:17.367] [nioEventLoopGroup-4-1] [c.g.h.r.c.h.RpcClientHandler.channelRead0] - [Client] response is :DefaultRpcResponse{seqId='ac246a40535e41ddb27768be80e5b381', error=null, result=CalculateResponse{success=true, sum=30}}
    [INFO] [2019-11-01 17:38:17.367] [main] [c.g.h.r.c.i.i.DefaultInvokeService.getResponse] - [Client] seq ac246a40535e41ddb27768be80e5b381 对应结果已经获取: DefaultRpcResponse{seqId='ac246a40535e41ddb27768be80e5b381', error=null, result=CalculateResponse{success=true, sum=30}}
    CalculateResponse{success=true, sum=30}

# 更多思考 #

如果是企业内部的框架，可以让用户不关心 register address。

直接在配置中心进行配置，用户在使用的时候无感。