我是一段不羁的公告！
记得给艿艿这 3 个项目加油，添加一个 STAR 噢。
https://github.com/YunaiV/SpringBoot-Labs
https://github.com/YunaiV/onemall
https://github.com/YunaiV/ruoyi-vue-pro
Dubbo

精尽 Dubbo 源码分析 —— 服务调用（八）之远程调用（Redis）

本文基于 Dubbo 2.6.1 版本，望知悉。

1. 概述

本文，我们分享 redis:// 协议的远程调用，主要分成两个个部分：

  • 服务暴露
  • 服务引用
  • 服务调用

对应项目为 dubbo-rpc-redis 。

对应文档为 《Dubbo 用户指南 —— redis://》 。定义如下：

基于 Redis 实现的 RPC 协议。

简单的说，通过 Dubbo Service 的调用方式，透明化对 Redis 的访问。
这样，如果未来希望，修改缓存的解决方案，不用修改代码，而只要修改 Dubbo Service 的配置。
就好像，Java JDBC API 有 MySQL JDBC 、Oracle JDBC 等多种实现，只需要修改对应的 JDBC 驱动实现类，就可以连接上不同的数据库。

另外，Dubbo 提供 memcached:// 协议，和 redis:// 对等，差别点在前者使用 Memcached ，后者使用 Redis 。

2. RedisProtocol

com.alibaba.dubbo.rpc.protocol.redis.RedisProtocol ，实现 AbstractProtocol 抽象类，redis:// 协议实现类。

2.1 export

@Override
public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
    throw new UnsupportedOperationException("Unsupported export redis service. url: " + invoker.getUrl());
}

实际访问的就是 Redis Server 实例，因此无需进行 Dubbo 服务暴露。客户端配置引用方式如下：

在客户端使用，注册中心读取：
<dubbo:reference id="store" interface="java.util.Map" group="member" />

或者，点对点直连：
<dubbo:reference id="store" interface="java.util.Map" url="redis://10.20.153.10:6379"

2.2 refer

  1: @Override
  2: public <T> Invoker<T> refer(final Class<T> type, final URL url) throws RpcException {
  3:     try {
  4:         // 创建 GenericObjectPoolConfig 对象，设置配置
  5:         GenericObjectPoolConfig config = new GenericObjectPoolConfig();
  6:         config.setTestOnBorrow(url.getParameter("test.on.borrow", true));
  7:         config.setTestOnReturn(url.getParameter("test.on.return", false));
  8:         config.setTestWhileIdle(url.getParameter("test.while.idle", false));
  9:         if (url.getParameter("max.idle", 0) > 0)
 10:             config.setMaxIdle(url.getParameter("max.idle", 0));
 11:         if (url.getParameter("min.idle", 0) > 0)
 12:             config.setMinIdle(url.getParameter("min.idle", 0));
 13:         if (url.getParameter("max.active", 0) > 0)
 14:             config.setMaxTotal(url.getParameter("max.active", 0));
 15:         if (url.getParameter("max.total", 0) > 0)
 16:             config.setMaxTotal(url.getParameter("max.total", 0));
 17:         if (url.getParameter("max.wait", 0) > 0)
 18:             config.setMaxWaitMillis(url.getParameter("max.wait", 0));
 19:         if (url.getParameter("num.tests.per.eviction.run", 0) > 0)
 20:             config.setNumTestsPerEvictionRun(url.getParameter("num.tests.per.eviction.run", 0));
 21:         if (url.getParameter("time.between.eviction.runs.millis", 0) > 0)
 22:             config.setTimeBetweenEvictionRunsMillis(url.getParameter("time.between.eviction.runs.millis", 0));
 23:         if (url.getParameter("min.evictable.idle.time.millis", 0) > 0)
 24:             config.setMinEvictableIdleTimeMillis(url.getParameter("min.evictable.idle.time.millis", 0));
 25:         // 创建 JedisPool 对象
 26:         final JedisPool jedisPool = new JedisPool(config, url.getHost(), url.getPort(DEFAULT_PORT),
 27:                 url.getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
 28: 
 29:         // 处理方法名的映射
 30:         final int expiry = url.getParameter("expiry", 0);
 31:         final String get = url.getParameter("get", "get");
 32:         final String set = url.getParameter("set", Map.class.equals(type) ? "put" : "set");
 33:         final String delete = url.getParameter("delete", Map.class.equals(type) ? "remove" : "delete");
 34: 
 35:         // 创建 Invoker 对象
 36:         return new AbstractInvoker<T>(type, url) {
 37: 
 38:             @Override
 39:             protected Result doInvoke(Invocation invocation) {
 40:                 Jedis resource = null;
 41:                 try {
 42:                     // 获得 Redis Resource
 43:                     resource = jedisPool.getResource();
 44:                     // Redis get 指令
 45:                     if (get.equals(invocation.getMethodName())) {
 46:                         if (invocation.getArguments().length != 1) {
 47:                             throw new IllegalArgumentException("The redis get method arguments mismatch, must only one arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url);
 48:                         }
 49:                         // 获得值
 50:                         byte[] value = resource.get(String.valueOf(invocation.getArguments()[0]).getBytes());
 51:                         if (value == null) {
 52:                             return new RpcResult();
 53:                         }
 54:                         // 反序列化
 55:                         ObjectInput oin = getSerialization(url).deserialize(url, new ByteArrayInputStream(value));
 56:                         // 返回结果
 57:                         return new RpcResult(oin.readObject());
 58:                     // Redis set/put 指令
 59:                     } else if (set.equals(invocation.getMethodName())) {
 60:                         if (invocation.getArguments().length != 2) {
 61:                             throw new IllegalArgumentException("The redis set method arguments mismatch, must be two arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url);
 62:                         }
 63:                         // 序列化
 64:                         byte[] key = String.valueOf(invocation.getArguments()[0]).getBytes();
 65:                         ByteArrayOutputStream output = new ByteArrayOutputStream();
 66:                         ObjectOutput value = getSerialization(url).serialize(url, output);
 67:                         value.writeObject(invocation.getArguments()[1]);
 68:                         // 设置值
 69:                         resource.set(key, output.toByteArray());
 70:                         if (expiry > 1000) {
 71:                             resource.expire(key, expiry / 1000);
 72:                         }
 73:                         // 返回结果
 74:                         return new RpcResult();
 75:                     } else if (delete.equals(invocation.getMethodName())) {
 76:                         if (invocation.getArguments().length != 1) {
 77:                             throw new IllegalArgumentException("The redis delete method arguments mismatch, must only one arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url);
 78:                         }
 79:                         // 删除值
 80:                         resource.del(String.valueOf(invocation.getArguments()[0]).getBytes());
 81:                         // 返回结果
 82:                         return new RpcResult();
 83:                     } else {
 84:                         throw new UnsupportedOperationException("Unsupported method " + invocation.getMethodName() + " in redis service.");
 85:                     }
 86:                 } catch (Throwable t) {
 87:                     RpcException re = new RpcException("Failed to invoke redis service method. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url + ", cause: " + t.getMessage(), t);
 88:                     if (t instanceof TimeoutException || t instanceof SocketTimeoutException) {
 89:                         re.setCode(RpcException.TIMEOUT_EXCEPTION);
 90:                     } else if (t instanceof JedisConnectionException || t instanceof IOException) {
 91:                         re.setCode(RpcException.NETWORK_EXCEPTION);
 92:                     } else if (t instanceof JedisDataException) {
 93:                         re.setCode(RpcException.SERIALIZATION_EXCEPTION);
 94:                     }
 95:                     throw re;
 96:                 } finally {
 97:                     // 归还 Redis Resource
 98:                     if (resource != null) {
 99:                         try {
100:                             jedisPool.returnResource(resource);
101:                         } catch (Throwable t) {
102:                             logger.warn("returnResource error: " + t.getMessage(), t);
103:                         }
104:                     }
105:                 }
106:             }
107: 
108:             @Override
109:             public void destroy() {
110:                 // 标记销毁
111:                 super.destroy();
112:                 // 销毁 Redis Pool
113:                 try {
114:                     jedisPool.destroy();
115:                 } catch (Throwable e) {
116:                     logger.warn(e.getMessage(), e);
117:                 }
118:             }
119: 
120:         };
121:     } catch (Throwable t) {
122:         throw new RpcException("Failed to refer redis service. interface: " + type.getName() + ", url: " + url + ", cause: " + t.getMessage(), t);
123:     }
124: }
  • 使用 Jedis 访问 Redis Server 。
  • 第 4 至 24 行：创建 GenericObjectPoolConfig 对象，从 Dubbo URL 中读取相关配置。此时，我们可以看到，为什么 Dubbo 的配置类中，有 arguments 属性了。可以使用它，实现不同 Protocol 协议的自定义属性。
  • 第 25 至 27 行：创建 JedisPool 对象。

  • 第 29 至 33 行：处理方法名的映射。

    如果方法名和 redis 的标准方法名不相同，则需要配置映射关系：
    <dubbo:reference id="cache" interface="com.foo.CacheService" url="memcached://10.20.153.10:11211" p:set="putFoo" p:get="getFoo" p:delete="removeFoo" />

    • 当对应的服务接口是 java.util.Map 时，对应的 Redis 数据结构为 Map 。

  • 第 35 至 120 行：创建 Invoker 对象。

2.2.1 doInvoke

  • 第 43 行：获得 Redis Resource 对象。
  • 第 44 至 57 行：Redis get 指令。
  • 第 58 至 74 行：Redis set/put 指令。
  • 第 75 至 83 行：Redis delete/remove 指令。
  • 第 84 至 85 行：目前其他命令，暂时不支持。
  • 第 86 至 95 行：翻译异常成 Dubbo 错误码。
  • 第 97 至 105 行：归还 Redis Resource 对象。

2.2.2 destroy

  • 第 111 行：调用 super#destroy() 方法，标记销毁。
  • 第 112 至 117 行：调用 JedisPool#destroy() 方法，销毁 Redis Pool 。

666. 彩蛋

知识星球知识星球

没有彩蛋~

总访客数 3341172 次 && 总访问量 6502571 次