Dubbo源码学习--redis协议（六）

╰+攻爆jí腚メ 2022-05-27 08:15 174阅读 0赞

基于 Redis [1][] 实现的 RPC 协议 [2][]。

## 注册 redis 服务的地址 ##

RegistryFactory registryFactory = ExtensionLoader.getExtensionLoader(RegistryFactory.class).getAdaptiveExtension();
    Registry registry = registryFactory.getRegistry(URL.valueOf("zookeeper://10.20.153.10:2181"));
    registry.register(URL.valueOf("redis://10.20.153.11/com.foo.BarService?category=providers&dynamic=false&application=foo&group=member&loadbalance=consistenthash"));

## 在客户端引用 ##

在客户端使用 [3][]：

<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" />

也可以使用自定义接口：

<dubbo:reference id="store" interface="com.foo.StoreService" url="redis://10.20.153.10:6379" />

方法名建议和 redis 的标准方法名相同，即：get(key), set(key, value), delet(key)。

如果方法名和 redis 的标准方法名不相同，则需要配置映射关系 [4][]：

<dubbo:reference id="cache" interface="com.foo.CacheService" url="memcached://10.20.153.10:11211" p:set="putFoo" p:get="getFoo" p:delete="removeFoo" />

> 1.  [Redis][] 是一个高效的 KV 存储服务器 [ ↩][Link 1]

> 2.  `2.3.0` 以上版本支持 [ ↩][Link 2]

> 3. 不需要感知 Redis 的地址 [ ↩][Link 3]

> 4. 其中 "p:xxx" 为 spring 的标准 p 标签 [ ↩][Link 4]

在RedisProtocol中同样提供了export和refer两个方法

（1）export方法：目前redis协议不支持对外提供服务的操作

（2）refer方法：在refer方法中redisProtocol完成了数据的设置、查询和删除操作，这样是否几个服务消费者就可以共享数据。

redis协议中没有服务提供者，简单来说更像是消费者将增删数据放到redis中，这样消费者自己及其他消费者可以通过接口进行获取数据，并设置数据的超时时间。

public class RedisProtocol extends AbstractProtocol {
    
        public static final int DEFAULT_PORT = 6379;
    
        public int getDefaultPort() {
            return DEFAULT_PORT;
        }
    
    	//redis协议无法暴露服务
        public <T> Exporter<T> export(final Invoker<T> invoker) throws RpcException {
            throw new UnsupportedOperationException("Unsupported export redis service. url: " + invoker.getUrl());
        }
    
        private Serialization getSerialization(URL url) {
            return ExtensionLoader.getExtensionLoader(Serialization.class).getExtension(url.getParameter(Constants.SERIALIZATION_KEY, "java"));
        }
    
    	//redis根据方法名称，完成向redis中添加数据、获取数据和删除数据操作
        public <T> Invoker<T> refer(final Class<T> type, final URL url) throws RpcException {
            try {
                GenericObjectPoolConfig config = new GenericObjectPoolConfig();
                config.setTestOnBorrow(url.getParameter("test.on.borrow", true));
                config.setTestOnReturn(url.getParameter("test.on.return", false));
                config.setTestWhileIdle(url.getParameter("test.while.idle", false));
                if (url.getParameter("max.idle", 0) > 0)
                    config.setMaxIdle(url.getParameter("max.idle", 0));
                if (url.getParameter("min.idle", 0) > 0)
                    config.setMinIdle(url.getParameter("min.idle", 0));
                if (url.getParameter("max.active", 0) > 0)
                    config.setMaxTotal(url.getParameter("max.active", 0));
                if (url.getParameter("max.total", 0) > 0)
                    config.setMaxTotal(url.getParameter("max.total", 0));
                if (url.getParameter("max.wait", 0) > 0)
                    config.setMaxWaitMillis(url.getParameter("max.wait", 0));
                if (url.getParameter("num.tests.per.eviction.run", 0) > 0)
                    config.setNumTestsPerEvictionRun(url.getParameter("num.tests.per.eviction.run", 0));
                if (url.getParameter("time.between.eviction.runs.millis", 0) > 0)
                    config.setTimeBetweenEvictionRunsMillis(url.getParameter("time.between.eviction.runs.millis", 0));
                if (url.getParameter("min.evictable.idle.time.millis", 0) > 0)
                    config.setMinEvictableIdleTimeMillis(url.getParameter("min.evictable.idle.time.millis", 0));
                final JedisPool jedisPool = new JedisPool(config, url.getHost(), url.getPort(DEFAULT_PORT),
                        url.getParameter(Constants.TIMEOUT_KEY, Constants.DEFAULT_TIMEOUT));
                final int expiry = url.getParameter("expiry", 0);
                final String get = url.getParameter("get", "get");
                final String set = url.getParameter("set", Map.class.equals(type) ? "put" : "set");
                final String delete = url.getParameter("delete", Map.class.equals(type) ? "remove" : "delete");
    			//创建代理实现类
                return new AbstractInvoker<T>(type, url) {
                    protected Result doInvoke(Invocation invocation) throws Throwable {
                        Jedis resource = null;
                        try {
                            resource = jedisPool.getResource();
    						//如果是get方法则从redis中获取数据
                            if (get.equals(invocation.getMethodName())) {
                                if (invocation.getArguments().length != 1) {
                                    throw new IllegalArgumentException("The redis get method arguments mismatch, must only one arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url);
                                }
                                byte[] value = resource.get(String.valueOf(invocation.getArguments()[0]).getBytes());
                                if (value == null) {
                                    return new RpcResult();
                                }
                                ObjectInput oin = getSerialization(url).deserialize(url, new ByteArrayInputStream(value));
                                return new RpcResult(oin.readObject());
    						//如果是set方法则向redis中添加数据
                            } else if (set.equals(invocation.getMethodName())) {
                                if (invocation.getArguments().length != 2) {
                                    throw new IllegalArgumentException("The redis set method arguments mismatch, must be two arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url);
                                }
                                byte[] key = String.valueOf(invocation.getArguments()[0]).getBytes();
                                ByteArrayOutputStream output = new ByteArrayOutputStream();
                                ObjectOutput value = getSerialization(url).serialize(url, output);
                                value.writeObject(invocation.getArguments()[1]);
                                resource.set(key, output.toByteArray());
                                if (expiry > 1000) {
                                    resource.expire(key, expiry / 1000);
                                }
                                return new RpcResult();
    						//如果是delete方法则删除redis中相关数据
                            } else if (delete.equals(invocation.getMethodName())) {
                                if (invocation.getArguments().length != 1) {
                                    throw new IllegalArgumentException("The redis delete method arguments mismatch, must only one arguments. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url);
                                }
                                resource.del(String.valueOf(invocation.getArguments()[0]).getBytes());
                                return new RpcResult();
                            } else {
                                throw new UnsupportedOperationException("Unsupported method " + invocation.getMethodName() + " in redis service.");
                            }
                        } catch (Throwable t) {
                            RpcException re = new RpcException("Failed to invoke redis service method. interface: " + type.getName() + ", method: " + invocation.getMethodName() + ", url: " + url + ", cause: " + t.getMessage(), t);
                            if (t instanceof TimeoutException || t instanceof SocketTimeoutException) {
                                re.setCode(RpcException.TIMEOUT_EXCEPTION);
                            } else if (t instanceof JedisConnectionException || t instanceof IOException) {
                                re.setCode(RpcException.NETWORK_EXCEPTION);
                            } else if (t instanceof JedisDataException) {
                                re.setCode(RpcException.SERIALIZATION_EXCEPTION);
                            }
                            throw re;
                        } finally {
                            if (resource != null) {
                                try {
                                    jedisPool.returnResource(resource);
                                } catch (Throwable t) {
                                    logger.warn("returnResource error: " + t.getMessage(), t);
                                }
                            }
                        }
                    }
    
                    public void destroy() {
                        super.destroy();
                        try {
                            jedisPool.destroy();
                        } catch (Throwable e) {
                            logger.warn(e.getMessage(), e);
                        }
                    }
                };
            } catch (Throwable t) {
                throw new RpcException("Failed to refer redis service. interface: " + type.getName() + ", url: " + url + ", cause: " + t.getMessage(), t);
            }
        }
    
    }

[1]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#fn_1
[2]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#fn_2
[3]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#fn_3
[4]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#fn_4
[Redis]: http://redis.io/
[Link 1]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#reffn_1
[Link 2]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#reffn_2
[Link 3]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#reffn_3
[Link 4]: http://dubbo.apache.org/books/dubbo-user-book/references/protocol/redis.html#reffn_4