RedissonV1.0源码分析-分布式对象篇

这篇主要分析的是 Redisson V1.0 版本分布式对象的实现，也是最为核心的源码分析。

这个版本的 Redisson 一共提供了下面几个分布式对象：

• 分布式 AtomicLong

• 分布式 CountDownLatch

• 分布式 List

• 分布式 Map

• 分布式 Queue

• 分布式 Set

• 分布式 Topic

• 分布式 Lock

可以看出 V1.0 版本这些分布式对象基本可以满足实际项目中大部分的场景需求。

Redisson 中每一个分布式对象都有各自的接口抽象，从接口的定义就可以直接看出来接口可以实现什么功能。

1. 分布式对象基类

1.1 RObject 接口定义

这个接口是 Redisson 中所有分布式对象的基础接口。

package org.redisson.core;


public interface RObject {

    // 获取对象的名称
    String getName();

}

1.2 RedissonObject 抽象类定义

RedissonObject 是一个基础的抽象类，不能被直接实例化，只能用作其他类的基类。

package org.redisson;

abstract class RedissonObject {

    private String name;  // 分布式对象的名称，用于 Redis 里面存储

    public RedissonObject(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    public void close() {
    }

}

2. 分布式 AtomicLong

分布式 AtomicLong 对象主要是利用了 Redis 的元素值自增、递减等特性。

2.1 RAtomicLong 接口定义

Java 标准库中 AtomicLong 的分布式替代版本，跟 AtomicLong 的功能几乎一样。

package org.redisson.core;

public interface RAtomicLong extends RObject {

    // 减 1, 再返回减之前的值
    long getAndDecrement();

    // 加指定数值，再返回加之后的值
    long addAndGet(long delta);

    // 如果当前的值等于预期的值，原子性地设置为新的值。 返回值是操作是否成功
    boolean compareAndSet(long expect, long update);

    // 减 1, 再返回减之后的值
    long decrementAndGet();

    // 获取当前值
    long get();

    // 加指定数值，再返回加之前的值
    long getAndAdd(long delta);

    // 设置新的值，返回旧的值
    long getAndSet(long newValue);

    // 自增1，然后返回自增后的值
    long incrementAndGet();

    // 自增1，然后返回自增前的值
    long getAndIncrement();

    // 设置新的值
    void set(long newValue);

}

2.2 RedissonAtomicLong 源码实现

分布式 AtomicLong 的实现。

package org.redisson;

import org.redisson.connection.ConnectionManager;
import org.redisson.core.RAtomicLong;

import com.lambdaworks.redis.RedisConnection;


public class RedissonAtomicLong extends RedissonObject implements RAtomicLong {

    private final ConnectionManager connectionManager;

    RedissonAtomicLong(ConnectionManager connectionManager, String name) {
        super(name);
        this.connectionManager = connectionManager;
    }


    @Override
    public long addAndGet(long delta) {
    	// 获取一个可用连接
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 调用 incrby 方法，实际上调用的是 INCRBY 指令
            return conn.incrby(getName(), delta);
        } finally {
        	// 释放当前连接
            connectionManager.release(conn);
        }
    }

    @Override
    public boolean compareAndSet(long expect, long update) {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 无限循环，直到设置成功为止
            while (true) {
            	// 使用 Redis 的 WATCH 命令监视 key
                conn.watch(getName());

                // 如果当前值不等于期望值，放弃事务并直接返回 false
                Long value = (Long) conn.get(getName());
                if (value != expect) {
                    conn.discard();
                    return false;
                }

                // 开始事务、设置新的值
                conn.multi();
                conn.set(getName(), update);

                // 返回结果大小为1 则说明事务成功，返回 true。
                if (conn.exec().size() == 1) {
                    return true;
                }
            }
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public long decrementAndGet() {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 调用 decr 方法，实际上调用的是 Redis 的 DECR 指令        	
            return conn.decr(getName());
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public long get() {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 调用 decr 方法，实际上调用的是 Redis 的 GET 指令        	
            return (Long) conn.get(getName());
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public long getAndAdd(long delta) {
        while (true) {
            long current = get();
            long next = current + delta;
            if (compareAndSet(current, next))
                return current;
        }
    }

    @Override
    public long getAndSet(long newValue) {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 实际上调用的是 Redis 的 GETSET 指令        	
            return (Long) conn.getset(getName(), newValue);
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public long incrementAndGet() {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 调用 incr 方法，实际上调用的是 Redis 的 INCR 指令        	
            return conn.incr(getName());
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public long getAndIncrement() {
        return getAndAdd(1);
    }

    public long getAndDecrement() {
        return getAndAdd(-1);
    }

    @Override
    public void set(long newValue) {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 调用 Redis SET 指令设置值
            conn.set(getName(), newValue);
        } finally {
            connectionManager.release(conn);
        }
    }

    public String toString() {
        return Long.toString(get());
    }

}

3. 分布式 CountDownLatch

分布式 CountDownLatch 主要是利用了 Redis 的 get、set 、发布订阅机制实现。

3.1 RCountDownLatch 接口定义

这个就是相当于 Java 标准库里 java.util.concurrent.CountDownLatch 的分布式替代版本。

比如 java.util.concurrent.CountDownLatch 更有优势的地方在于计数值可以通过 trySetCount 方法重置。

运用在多个 JVM 或服务器之间去共享或者同步状态。

package org.redisson.core;

import java.util.concurrent.TimeUnit;

public interface RCountDownLatch extends RObject {

    // 方法会导致当前线程等待，直到计数器降到零。除非是当前线程被中断。
    // 计数器为零，方法会立即返回。
    void await() throws InterruptedException;

    // 支持超时时间的等待。
    boolean await(long timeout, TimeUnit unit) throws InterruptedException;

    // 计数器减1
    void countDown();

    // 获取当前计数器的值。 用于调试跟测试
    long getCount();


    // 在计数器到0或者还没有设置的时候，设置一个新的计数器值。
    /  如果之前设置成功了计数值则返回 true
    /  如果之前的计数值还没有到达到 0 就返回 false
    boolean trySetCount(long count);

}

3.2 RedissonCountDownLatch 源码实现

Java java.util.concurrent.CountDownLatch 的分布式实现。

package org.redisson;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

import org.redisson.connection.ConnectionManager;
import org.redisson.connection.ConnectionManager.PubSubEntry;
import org.redisson.core.RCountDownLatch;
import org.redisson.misc.ReclosableLatch;

import com.lambdaworks.redis.RedisConnection;
import com.lambdaworks.redis.pubsub.RedisPubSubAdapter;


public class RedissonCountDownLatch extends RedissonObject implements RCountDownLatch {

    private final CountDownLatch subscribeLatch = new CountDownLatch(1);  // 确保订阅操作只执行一次

    private final String groupName = "redisson_countdownlatch_";  // 构建 Redis 中的 channel 名称的前缀

    private static final Integer zeroCountMessage = 0;  // 发布/订阅模式中传递消息的常量
    private static final Integer newCountMessage = 1;

    private final AtomicBoolean subscribeOnce = new AtomicBoolean();  // 确保只订阅一次

    private final ReclosableLatch msg = new ReclosableLatch();  // 可重复的闭锁，用于线程间的同步
 
    private final ConnectionManager connectionManager;  // 连接管理
    private PubSubEntry pubSubEntry;   // 发布订阅的对象

    RedissonCountDownLatch(ConnectionManager connectionManager, String name) {
        super(name);
        this.connectionManager = connectionManager;
    }

    public void subscribe() {
    	// 防止多线程环境下重复订阅
        if (subscribeOnce.compareAndSet(false, true)) {
            RedisPubSubAdapter<String, Integer> listener = new RedisPubSubAdapter<String, Integer>() {

                @Override
                public void subscribed(String channel, long count) {
                	// 查订阅的 channel 是否是期望的。是预期的调用 subscribeLatch.countDown()，释放等待订阅完成的线程
                    if (getChannelName().equals(channel)) {
                        subscribeLatch.countDown();
                    }
                }

                @Override
                public void message(String channel, Integer message) {
                    if (!getChannelName().equals(channel)) {
                        return;
                    }
					// 收到值为0 表示计数已经到达零，调用 msg.open() 来释放等待的线程
                    if (message.equals(zeroCountMessage)) {
                        msg.open();
                    }

                   // 收到值为 1 表示设置了新的计数，调用 msg.close() 来重置等待状态
                    if (message.equals(newCountMessage)) {
                        msg.close();
                    }
                }

            };
            // 订阅指定的 channel，并关联创建的监听器
            pubSubEntry = connectionManager.subscribe(listener, getChannelName());
        }

        // 等待订阅操作完成。await() 会阻塞当前线程，直到 countDown() 被调用
        try {
            subscribeLatch.await();
        } catch (InterruptedException e) {
        	// 等待过程中被中断，重新设置线程的中断状态
            Thread.currentThread().interrupt();
        }
    }

    public void await() throws InterruptedException {
    	// 循环会一直执行，直到计数器的值变为 0 或小于 0
    	// 这里会循环通过访问 Redis 来获取最新的计数值

        while (getCount() > 0) {
            // waiting for message
            msg.await();
        }
    }


    // 带超时时间的等待
    @Override
    public boolean await(long time, TimeUnit unit) throws InterruptedException {
        time = unit.toMillis(time);
        while (getCount() > 0) {
            if (time <= 0) {
                return false;
            }
            // 记录下当前的时间
            long current = System.currentTimeMillis();
            // 等待消息，传入超时时间等待
            msg.await(time, TimeUnit.MILLISECONDS);

            // 扣减本次用掉的时间
            long elapsed = System.currentTimeMillis() - current;
            time = time - elapsed;
        }

        return true;
    }

    @Override
    public void countDown() {
    	// 当前计数已经小于等于 0 直接返回
        if (getCount() <= 0) {
            return;
        }


        RedisConnection<String, Object> connection = connectionManager.connection();
        try {
        	// 调用 Redis 指令对数值减 1
            Long val = connection.decr(getName());

            // 等于 0 的操作
            if (val == 0) {
            	// 开启事务 > 删除计数器  > 发布消息通知等待的其他线程  >  执行事务检查操作结果
                connection.multi();
                connection.del(getName());
                connection.publish(getChannelName(), zeroCountMessage);
                if (connection.exec().size() != 2) {
                    throw new IllegalStateException();
                }
            } else if (val < 0) {
            	// 小于 0 是异常情况，直接删除掉
                connection.del(getName());
            }
        } finally {
            connectionManager.release(connection);
        }
    }

    private String getChannelName() {
        return groupName + getName();
    }

    // 获取计数器当前值
    @Override
    public long getCount() {
        RedisConnection<String, Object> connection = connectionManager.connection();
        try {
        	// 直接调用 Redis get 指令查询计数器的当前值, 查询不到默认为 0 
            Number val = (Number) connection.get(getName());
            if (val == null) {
                return 0;
            }
            return val.longValue();
        } finally {
            connectionManager.release(connection);
        }
    }

    // 尝试设置计数器的值
    @Override
    public boolean trySetCount(long count) {

        RedisConnection<String, Object> connection = connectionManager.connection();
        try {
        	// 使用 WATCH 命令监视计数器键，为后续的事务做准备
            connection.watch(getName());

            // 获取这个名称的计数器是不是有值，有值就放弃本次操作
            Long oldValue = (Long) connection.get(getName());
            if (oldValue != null) {
                connection.discard();
                return false;
            }

            开始事务 > 设置计数器的值到 Redis > 发布一个消息通知其他线程有新的计数值设置 > 执行事务检查结果
            connection.multi();
            connection.set(getName(), count);
            connection.publish(getChannelName(), newCountMessage);
            return connection.exec().size() == 2;
        } finally {
            connectionManager.release(connection);
        }
    }

    public void close() {
        connectionManager.unsubscribe(pubSubEntry, getChannelName());
    }

}

4. 分布式 List

分布式 List 对象主要是利用了 Redis 的 List 类型存储数据。

4.1 RList 接口定义

这接口就是内置的 List 接口的分布式实现，没很特殊的地方。

package org.redisson.core;
import java.util.List;

public interface RList<V> extends List<V>, RObject {
}

4.2 RedissonList 源码实现

java.util.List 接口的分布式实现，主要是利用了 Redis 的 List 结构。

package org.redisson;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.NoSuchElementException;

import org.redisson.connection.ConnectionManager;
import org.redisson.core.RList;

import com.lambdaworks.redis.RedisConnection;


public class RedissonList<V> extends RedissonObject implements RList<V> {

    private int batchSize = 50;   // 批量操作的大小  默认为50。 这个参数会影响一些批量操作的性能

    private final ConnectionManager connectionManager;

    RedissonList(ConnectionManager connectionManager, String name) {
        super(name);
        this.connectionManager = connectionManager;
    }

    protected ConnectionManager getConnectionManager() {
        return connectionManager;
    }

    @Override
    public int size() {
        RedisConnection<String, Object> connection = connectionManager.connection();
        try {
        	// 调用 Redis LLEN 命令获取 List 的长度
            return connection.llen(getName()).intValue();
        } finally {
            connectionManager.release(connection);
        }
    }

    @Override
    public boolean isEmpty() {
        return size() == 0;
    }

    @Override
    public boolean contains(Object o) {
        return indexOf(o) != -1;
    }

    @Override
    public Iterator<V> iterator() {
        return listIterator();
    }

    @Override
    public Object[] toArray() {
        List<V> list = subList(0, size());
        return list.toArray();
    }

    @Override
    public <T> T[] toArray(T[] a) {
        List<V> list = subList(0, size());
        return list.toArray(a);
    }

    @Override
    public boolean add(V e) {
        return addAll(Collections.singleton(e));
    }

    @Override
    public boolean remove(Object o) {
        RedisConnection<String, Object> connection = connectionManager.connection();
        try {
        	// 使用 Redis LREM 移除目标元素。
            return connection.lrem(getName(), 1, o) > 0;
        } finally {
            connectionManager.release(connection);
        }
    }

    @Override
    public boolean containsAll(Collection<?> c) {
    	// 发送 Redis 命令查询是否非空
        if (isEmpty()) {
            return false;
        }

        RedisConnection<String, Object> connection = connectionManager.connection();
        try {
        	// 创建集合的副本
            Collection<Object> copy = new ArrayList<Object>(c);

            // 计算需要分多少批操作
            int to = div(size(), batchSize);
            for (int i = 0; i < to; i++) {
            	// 调用 LRANGE 命令获取一批元素
                List<Object> range = connection.lrange(getName(), i*batchSize, i*batchSize + batchSize - 1);

                // 迭代需要判断是否存在的元素
                for (Iterator<Object> iterator = copy.iterator(); iterator.hasNext();) {
                    Object obj = iterator.next();
                    // 看元素是不是在本次 Redis 查询出来的数据中
                    int index = range.indexOf(obj);
                    // 找到了从副本集合中移除
                    if (index != -1) {
                        iterator.remove();
                    }
                }
            }
            // 复制集合为空，说明所有元素都被找到
            return copy.isEmpty();
        } finally {
            connectionManager.release(connection);
        }

    }

    @Override
    public boolean addAll(Collection<? extends V> c) {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 调用 Redis RPUSH  将集合中的所有元素添加到 List 的末尾
            conn.rpush(getName(), c.toArray());
            return true;
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public boolean addAll(int index, Collection<? extends V> coll) {
    	// 先检查下标是不是超过 redis list 总元素大小了
        checkPosition(index);

        // 如果index小于当前列表的大小，则需要在中间插入元素
        if (index < size()) {
            RedisConnection<String, Object> conn = connectionManager.connection();
            try {
                while (true) {
                	// WATCH命令监视列表的变化
                    conn.watch(getName());
                    // LRANGE 获取从 index 到 List 末尾的所有元素
                    List<Object> tail = conn.lrange(getName(), index, size());
                    // 开启事务
                    conn.multi();
                    
                    // LTRIM 命令保留从开始到 index-1 的元素，移除 index 及其后的元素
                    conn.ltrim(getName(), 0, index - 1);
                    
                    // RPUSH命令将集合入参中的元素插入到 List 中
                    conn.rpush(getName(), coll.toArray());
                    
                    // 再次使用 RPUSH 命令将之前获取的尾部元素重新添加到列表中
                    conn.rpush(getName(), tail.toArray());

                    // 执行事务。 返回的结果大小为3则为执行成功  
                    if (conn.exec().size() == 3) {
                        return true;
                    }
                }
            } finally {
                connectionManager.release(conn);
            }
        } else {
        	// 下标在 List 末尾的情况
            return addAll(coll);
        }
    }

    @Override
    public boolean removeAll(Collection<?> c) {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
            boolean result = false;
            // 使用 LREM 命令逐个移除
            for (Object object : c) {
                boolean res = conn.lrem(getName(), 0, object) > 0;
                if (!result) {
                    result = res;
                }
            }
            return result;
        } finally {
            connectionManager.release(conn);
        }

    }

    // 保留传入集合中的元素，其他的删除掉
    @Override
    public boolean retainAll(Collection<?> c) {
        boolean changed = false;
        // 通过迭代器遍历 List，移除不在集合中的元素
        for (Iterator<V> iterator = iterator(); iterator.hasNext();) {
            V object = iterator.next();
            if (!c.contains(object)) {
                iterator.remove();
                changed = true;
            }
        }
        return changed;
    }

    @Override
    public void clear() {
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// DEL 命令删除元素
            conn.del(getName());
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public V get(int index) {
    	// 先检查下标是不是超过 redis list 总元素大小

        checkIndex(index);
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {

        	// LINDEX 命令获取指定索引的元素
            return (V) conn.lindex(getName(), index);
        } finally {
            connectionManager.release(conn);
        }
    }

    private void checkIndex(int index) {
        int size = size();
        if (!isInRange(index, size))
            throw new IndexOutOfBoundsException("index: " + index + " but current size: "+ size);
    }

    private boolean isInRange(int index, int size) {
        return index >= 0 && index < size;
    }

    private void checkPosition(int index) {
        int size = size();
        if (!isPositionInRange(index, size))
            throw new IndexOutOfBoundsException("index: " + index + " but current size: "+ size);
    }

    private boolean isPositionInRange(int index, int size) {
        return index >= 0 && index <= size;
    }


    @Override
    public V set(int index, V element) {
    	// 先检查下标是不是超过 redis list 总元素大小
        checkIndex(index);
        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// LINDEX 获取之前的对象
            V prev = (V) conn.lindex(getName(), index);
            // LSET 设置值  指定下标
            conn.lset(getName(), index, element);
            return prev;
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public void add(int index, V element) {
        addAll(index, Collections.singleton(element));
    }

    // 计算要处理的总批次
    private int div(int p, int q) {
        int div = p / q;
        int rem = p - q * div; // equal to p % q

        if (rem == 0) {
          return div;
        }

        return div + 1;
    }

    @Override
    public V remove(int index) {
        checkIndex(index);

        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
            if (index == 0) {
                return (V) conn.lpop(getName());
            }
            while (true) {
            	// WATCH 命令监视列表的变化
                conn.watch(getName());
                // LINDEX 命令获取指定下标元素
                V prev = (V) conn.lindex(getName(), index);

                // LRANG 取出从 index+1 到列表末尾的所有元素。
                List<Object> tail = conn.lrange(getName(), index + 1, size());

                // 开启事务
                conn.multi();
                // LTRIM 命令保留  Redis List 从开始到 index-1 的元素 。  移除index及其后的元素
                conn.ltrim(getName(), 0, index - 1);
                // RPUSH 命令将之前获取的尾部元素重新添加到 Redis List中
                conn.rpush(getName(), tail.toArray());
                // 返回的结果大小为 2 则为事务执行成功，返回上一个元素的值
                if (conn.exec().size() == 2) {
                    return prev;
                }
            }
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public int indexOf(Object o) {
        if (isEmpty()) {
            return -1;
        }

        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 计算总批次
            int to = div(size(), batchSize);
            for (int i = 0; i < to; i++) {
            	
            	// LRANG 命令取出一批数据，判断是否在这一批数据里面
                List<Object> range = conn.lrange(getName(), i*batchSize, i*batchSize + batchSize - 1);
                int index = range.indexOf(o);
                if (index != -1) {
                    return index + i*batchSize;
                }
            }

            return -1;
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public int lastIndexOf(Object o) {
        if (isEmpty()) {
            return -1;
        }

        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// 拿到 Redis List 总元素大小
            int size = size();
            // 计算要处理的总批次
            int to = div(size, batchSize);

            // 循环处理
            for (int i = 1; i <= to; i++) {
                int startIndex = -i*batchSize;

                // LRANGE命令获取当前批次的元素
                List<Object> range = conn.lrange(getName(), startIndex, size - (i-1)*batchSize);
                // 从 Redis 获取的数据范围内查找元素的最后一个索引
                int index = range.lastIndexOf(o);
                if (index != -1) {
                	// 找到了该元素，计算其在整个列表中的实际索引并返回
                    return Math.max(size + startIndex, 0) + index;
                }
            }

            return -1;
        } finally {
            connectionManager.release(conn);
        }
    }

    @Override
    public ListIterator<V> listIterator() {
        return listIterator(0);
    }

    // 迭代器实现
    // 支持双向遍历和修改

    @Override
    public ListIterator<V> listIterator(final int ind) {
        return new ListIterator<V>() {
        	// 当前迭代器指向的索引，初始化为ind - 1，因为 next() 会先增加索引
            private int currentIndex = ind - 1;
            // 是否执行过remove 操作，防止重复删除
            private boolean removeExecuted;

            @Override
            public boolean hasNext() {
            	// 检查是否存在下一个元素，确保 currentIndex+1 小于列表大小且列表非空
                int size = size();
                return currentIndex+1 < size && size > 0;
            }

            @Override
            public V next() {
            	// 返回下一个元素，如果没有下一个元素则抛出NoSuchElementException
                if (!hasNext()) {
                    throw new NoSuchElementException("No such element at index " + currentIndex);
                }
                // 更新 currentIndex 并重置 removeExecuted 状态
                currentIndex++;
                removeExecuted = false;
                return RedissonList.this.get(currentIndex);
            }

            @Override
            public void remove() {
            	// 已执行过remove()则抛出IllegalStateException。
                if (removeExecuted) {
                    throw new IllegalStateException("Element been already deleted");
                }
                // 调用  remove 方法   更新 currentIndex 和 removeExecuted
                RedissonList.this.remove(currentIndex);
                currentIndex--;
                removeExecuted = true;
            }

            @Override
            public boolean hasPrevious() {
                int size = size();
                return currentIndex-1 < size && size > 0 && currentIndex >= 0;
            }

            @Override
            public V previous() {
                if (!hasPrevious()) {
                    throw new NoSuchElementException("No such element at index " + currentIndex);
                }
                removeExecuted = false;
                V res = RedissonList.this.get(currentIndex);
                currentIndex--;
                return res;
            }

            @Override
            public int nextIndex() {
                return currentIndex + 1;
            }

            @Override
            public int previousIndex() {
                return currentIndex;
            }

            @Override
            public void set(V e) {
                if (currentIndex >= size()-1) {
                    throw new IllegalStateException();
                }
                RedissonList.this.set(currentIndex, e);
            }

            @Override
            public void add(V e) {
                RedissonList.this.add(currentIndex+1, e);
                currentIndex++;
            }
        };
    }

    @Override
    public List<V> subList(int fromIndex, int toIndex) {
        int size = size();
        if (fromIndex < 0 || toIndex > size) {
            throw new IndexOutOfBoundsException("fromIndex: " + fromIndex + " toIndex: " + toIndex + " size: " + size);
        }
        if (fromIndex > toIndex) {
            throw new IllegalArgumentException("fromIndex: " + fromIndex + " toIndex: " + toIndex);
        }

        RedisConnection<String, Object> conn = connectionManager.connection();
        try {
        	// LRANGE 命令获取指定范围的元素
            return (List<V>) conn.lrange(getName(), fromIndex, toIndex - 1);
        } finally {
            connectionManager.release(conn);
        }
    }

    public String toString() {
        Iterator<V> it = iterator();
        if (! it.hasNext())
            return "[]";

        StringBuilder sb = new StringBuilder();
        sb.append('[');
        for (;;) {
            V e = it.next();
            sb.append(e == this ? "(this Collection)" : e);
            if (! it.hasNext())
                return sb.append(']').toString();
            sb.append(',').append(' ');
        }
    }

}