@SPI("dubbo") public interface RegistryFactory { /** * 连接注册中心. * * 连接注册中心需处理契约： * 1. 当设置 check=false 时表示不检查连接，否则在连接不上时抛出异常。 * 2. 支持 URL 上的 username:password 权限认证。 * 3. 支持 backup=10.20.153.10 备选注册中心集群地址。 * 4. 支持 file=registry.cache 本地磁盘文件缓存。 * 5. 支持 timeout=1000 请求超时设置。 * 6. 支持 session=60000 会话超时或过期设置。 * @param url 注册中心地址，不允许为空 * @return 注册中心引用，总不返回空 */ @Adaptive({"protocol"}) Registry getRegistry(URL url); }

dubbo=com.alibaba.dubbo.registry.dubbo.DubboRegistryFactory multicast=com.alibaba.dubbo.registry.multicast.MulticastRegistryFactory zookeeper=com.alibaba.dubbo.registry.zookeeper.ZookeeperRegistryFactory redis=com.alibaba.dubbo.registry.redis.RedisRegistryFactory

public class ZookeeperRegistryFactory extends AbstractRegistryFactory { private ZookeeperTransporter zookeeperTransporter; public void setZookeeperTransporter(ZookeeperTransporter zookeeperTransporter) { this.zookeeperTransporter = zookeeperTransporter; } public Registry createRegistry(URL url) { return new ZookeeperRegistry(url, zookeeperTransporter); }}

public Registry getRegistry(URL url) { url = url.setPath(RegistryService.class.getName()) .addParameter(Constants.INTERFACE_KEY, RegistryService.class.getName()) .removeParameters(Constants.EXPORT_KEY, Constants.REFER_KEY); String key = url.toServiceString(); //锁定注册中心获取过程，保证注册中心单一实例 LOCK.lock(); try { Registry registry = REGISTRIES.get(key); if (registry != null) { return registry; } registry = createRegistry(url); if (registry == null) { throw new IllegalStateException("Can not create registry " + url); } REGISTRIES.put(key, registry); return registry; } finally { //释放锁 LOCK.unlock(); } }

public Registry createRegistry(URL url) { return new ZookeeperRegistry(url, zookeeperTransporter); }

public ZookeeperRegistry(URL url, ZookeeperTransporter zookeeperTransporter) { super(url); if (url.isAnyHost()) { throw new IllegalStateException("registry address == null"); } String group = url.getParameter(Constants.GROUP_KEY, DEFAULT_ROOT); if (! group.startsWith(Constants.PATH_SEPARATOR)) { group = Constants.PATH_SEPARATOR + group; } this.root = group; //设置根节点
// 建立连接 zkClient = zookeeperTransporter.connect(url); zkClient.addStateListener(new StateListener() { public void stateChanged(int state) { if (state == RECONNECTED) { try { recover(); } catch (Exception e) { logger.error(e.getMessage(), e); } } } }); }

public void register(URL url) { super.register(url); failedRegistered.remove(url); failedUnregistered.remove(url); try { // 向服务器端发送注册请求 doRegister(url); } catch (Exception e) { Throwable t = e; // 如果开启了启动时检测，则直接抛出异常 boolean check = getUrl().getParameter(Constants.CHECK_KEY, true) && url.getParameter(Constants.CHECK_KEY, true) && ! Constants.CONSUMER_PROTOCOL.equals(url.getProtocol()); boolean skipFailback = t instanceof SkipFailbackWrapperException; if (check || skipFailback) { if(skipFailback) { t = t.getCause(); } throw new IllegalStateException("Failed to register " + url + " to registry " + getUrl().getAddress() + ", cause: " + t.getMessage(), t); } else { logger.error("Failed to register " + url + ", waiting for retry, cause: " + t.getMessage(), t); }// 将失败的注册请求记录到失败列表，定时重试 failedRegistered.add(url); } }

protected void doRegister(URL url) { try {//调用 zkclient.create 在 zookeeper 中创建一个节点。 zkClient.create(toUrlPath(url), url.getParameter(Constants.DYNAMIC_KEY, true)); } catch (Throwable e) { throw new RpcException("Failed to register " + url + " to zookeeper " + getUrl() + ", cause: " + e.getMessage(), e); } }

Invoker invoker = proxyFactory.getInvoker(ref, (Class) interfaceClass, registryURL.addParameterAndEncoded(Constants.EXPORT_KEY, url.toFullString())); Exporter exporter = protocol.export(invoker);

public ExchangeServer bind(URL url, ExchangeHandler handler) throws RemotingException { return new HeaderExchangeServer(Transporters.bind(url, new DecodeHandler(new HeaderExchangeHandler(handler)))); }

public NettyServer(URL url, ChannelHandler handler) throws RemotingException { super(url, ChannelHandlers.wrap(handler, ExecutorUtil.setThreadName(url, SERVER_THREAD_POOL_NAME))); }

public static ChannelHandler wrap(ChannelHandler handler, URL url) { return ChannelHandlers.getInstance().wrapInternal(handler, url); }protected ChannelHandler wrapInternal(ChannelHandler handler, URL url) { return new MultiMessageHandler(new HeartbeatHandler(ExtensionLoader.getExtensionLoader(Dispatcher.class) .getAdaptiveExtension().dispatch(handler, url))); }

@Override public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception { NettyChannel channel = NettyChannel.getOrAddChannel(ctx.getChannel(), url, handler); try { handler.received(channel, e.getMessage()); } finally { NettyChannel.removeChannelIfDisconnected(ctx.getChannel()); } }

public void received(Channel channel, Object message) throws RemotingException { channel.setAttribute(KEY_READ_TIMESTAMP, System.currentTimeMillis()); ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel); try { if (message instanceof Request) {//进入这里 // handle request. Request request = (Request) message; if (request.isEvent()) { handlerEvent(channel, request); } else { if (request.isTwoWay()) {//双向请求/进入这里 Response response = handleRequest(exchangeChannel, request); channel.send(response); } else {// 单向 handler.received(exchangeChannel, request.getData()); } } 。。。。。。。。。 }

Response handleRequest(ExchangeChannel channel, Request req) throws RemotingException { Response res = new Response(req.getId(), req.getVersion()); if (req.isBroken()) { Object data = https://www.it610.com/article/req.getData(); String msg; if (data == null) msg = null; else if (data instanceof Throwable) msg = StringUtils.toString((Throwable) data); else msg = data.toString(); res.setErrorMessage("Fail to decode request due to: " + msg); res.setStatus(Response.BAD_REQUEST); return res; } // find handler by message class. Object msg = req.getData(); try { // handle data. Object result = handler.reply(channel, msg); res.setStatus(Response.OK); res.setResult(result); } catch (Throwable e) { res.setStatus(Response.SERVICE_ERROR); res.setErrorMessage(StringUtils.toString(e)); } return res; }

private ExchangeHandler requestHandler = new ExchangeHandlerAdapter() {

public Result invoke(Invocation invocation) throws RpcException { try { return new RpcResult(doInvoke(proxy, invocation.getMethodName(), invocation.getParameterTypes(), invocation.getArguments())); } catch (InvocationTargetException e) { return new RpcResult(e.getTargetException()); } catch (Throwable e) { throw new RpcException("Failed to invoke remote proxy method " + invocation.getMethodName() + " to " + getUrl() + ", cause: " + e.getMessage(), e); } }

推荐阅读

• 

  哺乳期可以打新型冠状病毒疫苗吗
• 

  12333社保卡余额查询 如何查询社保余额
• 

  浏阳漂流哪里最好玩最刺激？ 长沙浏阳漂流哪里最好玩最刺激
• 

  固元膏用什么黄酒怎么保存
• 

  商业健康保险主要包括
• 

  公司经营范围参考 注册公司
• 

  遇到一个处处看我不顺眼的领导怎么办？
• 

  字节跳动超越腾讯需要几年？
• 

  锦鲤尾巴上有白色点状物是咋回事
• 

  炖牛肉什么时候放白萝卜
• 

  嫩姜拌山参 野山参和生姜能一同下锅吗
• 

  衣服上的油漆怎么洗掉
• 

  佳能600d闪光灯怎么关 佳能600d闪光灯怎么开
• 

  白领怎么吃才能有效减压？
• 

  鸡腿菇的保存方法和温度 鸡腿菇的保存方法
• 

  铁木砧板初次使用保养用盐水泡多久
• 

  手机百度怎么删除书签
• 

  去日本必买的玩具 日本有什么好东西
• 

  抗过敏的食物有哪些？抗过敏的食物
• 

  惭怍什么意思 惭忸是什么意思

• Java|Java基础——数组
• 人工智能|干货！人体姿态估计与运动预测
• java简介|Java是什么（Java能用来干什么？）
• Java|规范的打印日志
• Linux|109 个实用 shell 脚本
• 程序员|【高级Java架构师系统学习】毕业一年萌新的Java大厂面经，最新整理
• Python专栏|数据分析的常规流程
• Spring注解驱动第十讲--@Autowired使用
• SqlServer|sql server的UPDLOCK、HOLDLOCK试验
• 读书笔记|《白话大数据和机器学习》学习笔记1