Question

10.1. 对 Redis 节点的操作

Redisson 的 NodesGroup 对象提供了许些对 Redis 节点的操作。

NodesGroup nodesGroup = redisson.getNodesGroup();
nodesGroup.addConnectionListener(new ConnectionListener() {
    public void onConnect(InetSocketAddress addr) {
       // Redis 节点连接成功
    }

    public void onDisconnect(InetSocketAddress addr) {
       // Redis 节点连接断开
    }
});

也可以用来 PING 单个 Redis 节点或全部节点。

NodesGroup nodesGroup = redisson.getNodesGroup();
Collection<Node> allNodes = nodesGroup.getNodes();
for (Node n : allNodes) {
    n.ping();
}
// 或者
nodesGroup.pingAll();

10.2. 复杂多维对象结构和对象引用的支持

Redisson 突破了 Redis 数据结构维度的限制，通过一个特殊引用对象的帮助，Redisson 允许以任意的组合方式构建多维度的复杂对象结构，实现了对象之间的类似传统数据库里的关联关系。使用范例如下：

RMap<RSet<RList>, RList<RMap>> map = redisson.getMap("myMap");
RSet<RList> set = redisson.getSet("mySet");
RList<RMap> list = redisson.getList("myList");

map.put(set, list);
// 在特殊引用对象的帮助下，我们甚至可以构建一个循环引用，这是通过普通序列化方式实现不了的。
set.add(list);
list.add(map);

可能您已经注意到了，在 map 包含的元素发生改变以后，我们无需再次“保存/持久”这些对象。因为 map 对象所记录的并不是序列化以后的值，而是元素对象的引用。这让 Redisson 提供的对象在使用方法上，与普通 Java 对象的使用方法一致。从而让 Redis 成为内存的一部分，而不仅仅是一个储存空间。

以上范例中，一共创建了三个 Redis 数据结构：一个 Redis HASH，一个 Redis SET 和一个 Redis LIST。

10.3. 命令的批量执行

多个连续命令可以通过 RBatch 对象在一次网络会话请求里合并发送，这样省去了产生多个请求消耗的时间和资源。这在 Redis 中叫做 管道 。

用户可以通过以下方式调整通过管道方式发送命令的方式：

BatchOptions options = BatchOptions.defaults()
// 指定执行模式
//
// ExecutionMode.REDIS_READ_ATOMIC - 所有命令缓存在 Redis 节点中，以原子性事务的方式执行。
//
// ExecutionMode.REDIS_WRITE_ATOMIC - 所有命令缓存在 Redis 节点中，以原子性事务的方式执行。
//
// ExecutionMode.IN_MEMORY - 所有命令缓存在 Redisson 本机内存中统一发送，但逐一执行（非事务）。默认模式。
//
// ExecutionMode.IN_MEMORY_ATOMIC - 所有命令缓存在 Redisson 本机内存中统一发送，并以原子性事务的方式执行。
//
.executionMode(ExecutionMode.IN_MEMORY)

// 告知 Redis 不用返回结果（可以减少网络用量）
.skipResult()

// 将写入操作同步到从节点
// 同步到 2 个从节点，等待时间为 1 秒钟
.syncSlaves(2, 1, TimeUnit.SECONDS)

// 处理结果超时为 2 秒钟
.responseTimeout(2, TimeUnit.SECONDS)

// 命令重试等待间隔时间为 2 秒钟
.retryInterval(2, TimeUnit.SECONDS);

// 命令重试次数。仅适用于未发送成功的命令
.retryAttempts(4);

使用方式如下：

RBatch batch = redisson.createBatch();
batch.getMap("test").fastPutAsync("1", "2");
batch.getMap("test").fastPutAsync("2", "3");
batch.getMap("test").putAsync("2", "5");
batch.getAtomicLongAsync("counter").incrementAndGetAsync();
batch.getAtomicLongAsync("counter").incrementAndGetAsync();

BatchResult res = batch.execute();
// 或者
Future<BatchResult> asyncRes = batch.executeAsync();
List<?> response = res.getResponses();
res.getSyncedSlaves();

在集群模式下，所有的命令会按各个槽所在的节点，筛选分配到各个节点并同时发送。每个节点返回的结果将会汇总到最终的结果列表里。

10.4. Redisson 事务

Redisson 为 RMap 、 RMapCache 、 RLocalCachedMap 、 RSet 、 RSetCache 和 RBucket 这样的对象提供了具有 ACID 属性的事务功能。Redisson 事务通过分布式锁保证了连续写入的原子性，同时在内部通过操作指令队列实现了 Redis 原本没有的 提交 与 滚回 功能。当 提交 与 滚回 遇到问题的时候，将通过 org.redisson.transaction.TransactionException 告知用户。

目前支持的环境如下： SINGLE , MASTER/SLAVE , SENTINEL , ELASTICACHE REPLICATED , AZURE CACHE , RLEC 。

Redisson 事务支持的事务隔离等级为: READ_COMMITTED ，即仅读取提交后的结果。

另见 Spring 事务管理器 和本章 XA 事务（XA Transactions） 。

以下选项可以用来配置事务属性：

TransactionOptions options = TransactionOptions.defaults()
// 设置参与本次事务的主节点与其从节点同步的超时时间。
// 默认值是 5 秒。
.syncSlavesTimeout(5, TimeUnit.SECONDS)

// 处理结果超时。
// 默认值是 3 秒。
.responseTimeout(3, TimeUnit.SECONDS)

// 命令重试等待间隔时间。仅适用于未发送成功的命令。
// 默认值是 1.5 秒。
.retryInterval(2, TimeUnit.SECONDS)

// 命令重试次数。仅适用于未发送成功的命令。
// 默认值是 3 次。
.retryAttempts(3)

// 事务超时时间。如果规定时间内没有提交该事务则自动滚回。
// 默认值是 5 秒。
.timeout(5, TimeUnit.SECONDS);

代码范例：

RTransaction transaction = redisson.createTransaction(TransactionOptions.defaults());

RMap<String, String> map = transaction.getMap("myMap");
map.put("1", "2");
String value = map.get("3");
RSet<String> set = transaction.getSet("mySet")
set.add(value);

try {
   transaction.commit();
} catch(TransactionException e) {
   transaction.rollback();
}

10.5. XA 事务（XA Transactions）

Redisson 提供了 XAResource 标准的实现。该实现可用于 JTA 事务中。

另见本章 Redisson 事务 和 Spring 事务管理器 。

该功能仅适用于 Redisson PRO 版本

代码范例：

// Transaction 对象可以从所有兼容 JTA 接口的事务管理器中获取。
Transaction globalTransaction = transactionManager.getTransaction();

RXAResource xaResource = redisson.getXAResource();
globalTransaction.enlistResource(xaResource);

RTransaction transaction = xaResource.getTransaction();
RBucket<String> bucket = transaction.getBucket("myBucket");
bucket.set("simple");
RMap<String, String> map = transaction.getMap("myMap");
map.put("myKey", "myValue");

transactionManager.commit();

10.6. 脚本执行

redisson.getBucket("foo").set("bar");
String r = redisson.getScript().eval(Mode.READ_ONLY,
   "return redis.call('get', 'foo')", RScript.ReturnType.VALUE);

// 通过预存的脚本进行同样的操作
RScript s = redisson.getScript();
// 首先将脚本保存到所有的 Redis 主节点
String res = s.scriptLoad("return redis.call('get', 'foo')");
// 返回值 res == 282297a0228f48cd3fc6a55de6316f31422f5d17

// 再通过 SHA 值调用脚本
Future<Object> r1 = redisson.getScript().evalShaAsync(Mode.READ_ONLY,
   "282297a0228f48cd3fc6a55de6316f31422f5d17",
   RScript.ReturnType.VALUE, Collections.emptyList());

10.7. 底层 Redis 客户端

Redisson 在底层采用了高性能异步非阻塞式 Java 客户端，它同时支持异步和同步两种通信模式。如果有哪些命令 Redisson 还没提供支持，也可以直接通过调用底层 Redis 客户端来实现。Redisson 支持的命令在 Redis 命令和 Redisson 对象匹配列表 里做了详细对比参照。

// 在使用多个客户端的情况下可以共享同一个 EventLoopGroup
EventLoopGroup group = new NioEventLoopGroup();

RedisClientConfig config = new RedisClientConfig();
config.setAddress("redis://localhost:6379") // 或者用 rediss://使用加密连接
      .setPassword("myPassword")
      .setDatabase(0)
      .setClientName("myClient")
      .setGroup(group);

RedisClient client = RedisClient.create(config);
RedisConnection conn = client.connect();
// 或
RFuture<RedisConnection> connFuture = client.connectAsync();

conn.sync(StringCodec.INSTANCE, RedisCommands.SET, "test", 0);
// 或
conn.async(StringCodec.INSTANCE, RedisCommands.GET, "test");

conn.close()
// 或
conn.closeAsync()

client.shutdown();
// 或
client.shutdownAsync();

280d386