水煮Redisson（二一）-批处理

前言

这一章介绍Redisson中的批处理类【RBatch】，功能上类似redis的管道操作，在同一个请求中，发送多条指令，节省网络I/O时间。 下面展示一个例子，测试用例中，一次发送了三条Map查询指令，然后同时返回。

    @Test
    public void testBatch() throws InterruptedException {
        // 创建redis链接
        RedissonClient redissonClient = create();
        // 创建批处理对象
        RBatch batch = redissonClient.createBatch(BatchOptions.defaults());
        // 批操作
        batch.getMap("TEST_MAP").getAsync("KEY_11");
        batch.getMap("TEST_MAP").getAsync("KEY_22");
        batch.getMap("TEST_MAP222").getAsync("KEY_33");
        // 异步执行
        RFuture<BatchResult<?>> executeAsync = batch.executeAsync();
        executeAsync.onComplete((r, e) -> {
            log.info("======================批处理结果:{}", r.getResponses());
        });
        // 等待结果
        executeAsync.await(5, TimeUnit.MINUTES);
    }

关键字

• CommandBatchService：批处理的入口
• RedisCommonBatchExecutor：批处理执行器，继承了【RedisExecutor】类，在上一章着重介绍过。
• commands：指令集合，类型为ConcurrentMap，MasterSlaveEntry是一个slot的映射，表示相同slot的指令，都会存放到同一个Entry中。上面的测试用例，有两个map key，假设这两个key分布在不同的slot中，commands集合就会有两个元素，图示如下：

添加指令

添加指令的入口是CommandBatchService.async(...)，判断操作对象是否在同一个slot，如果是的话，指令会放在一批发送，如果不是，则单独创建一个Entry，并放入到commands集合。后续执行和获取结果，都需要用到此集合。

protected final void addBatchCommandData(Object[] batchParams) {
    MasterSlaveEntry msEntry = getEntry(source);
    Entry entry = commands.get(msEntry);
	// 判断slot是否在批对象中存在，如果不存在，则需要创建
    if (entry == null) {
        commands.putIfAbsent(msEntry, new Entry());
    }
    Codec codecToUse = getCodec(codec);
	// 新建批指令
    BatchCommandData<V, R> commandData = new BatchCommandData<V, R>(mainPromise, codecToUse, command, batchParams, index.incrementAndGet());
	// 批指令加入到slot
    entry.getCommands().add(commandData);
}

批处理入口

执行方法的入口：CommandBatchService.executeAsync(...)，方法中触发执行和结果监听回传。

public RFuture<BatchResult<?>> executeAsync() {
    RPromise<BatchResult<?>> promise = new RedissonPromise<>();
    RPromise<Void> voidPromise = new RedissonPromise<Void>();
    // 结果监听
    voidPromise.onComplete((res, ex) -> {
        executed.set(true);
        // 整理指令
        List<BatchCommandData> entries = new ArrayList<BatchCommandData>();
        // commands的类型，ConcurrentMap<MasterSlaveEntry, Entry>，如果是相同slot，那么放在一批执行
        for (Entry e : commands.values()) {
            entries.addAll(e.getCommands());
        }
        // 排序，这里不太明白为什么要排序
        Collections.sort(entries);
        // 接收批处理结果
        List<Object> responses = new ArrayList<Object>(entries.size());
        int syncedSlaves = 0;
        for (BatchCommandData<?, ?> commandEntry : entries) {
            Object entryResult = commandEntry.getPromise().getNow();
            responses.add(entryResult);
        }
        // 结果组装
        BatchResult<Object> result = new BatchResult<Object>(responses, syncedSlaves);
        // 触发单个指令的成功事件，如果此promise注册了监听，则会立即触发
        promise.trySuccess(result);
    });
    // 每个指令，用RedisCommonBatchExecutor类单独执行
    for (Map.Entry<MasterSlaveEntry, Entry> e : commands.entrySet()) {
        // 相同slot，放在一批执行
        RedisCommonBatchExecutor executor = new RedisCommonBatchExecutor(new NodeSource(e.getKey()), voidPromise,
                                                connectionManager, this.options, e.getValue(), slots);
        executor.execute();
    }
    return promise;
}

执行指令

RedisCommonBatchExecutor：批处理执行器，继承了【RedisExecutor】类，主要重写了方法【sendCommand】，因为这里发送的是一个指令集合。

@Override
protected void sendCommand(RPromise<Void> attemptPromise, RedisConnection connection) {
    // 同一个slot中的指令集合
    List<CommandData<?, ?>> list = new ArrayList<>(entry.getCommands().size());
    // 集体发送到redis服务
    writeFuture = connection.send(new CommandsData(attemptPromise, list, options.isSkipResult(), isAtomic, isQueued, options.getSyncSlaves() > 0));
}

发送到redis服务

上面代码中，connection.send方法，后续会对指令进行编码【CommandBatchEncoder】。在下面展示的代码中，可以看到redisson中的批处理，没有使用redis服务的pipeline指令，而是通过遍历，将集合中的指令，分条进行编码，然后一起发送到redis服务。

    @Override
    protected void encode(ChannelHandlerContext ctx, CommandsData msg, ByteBuf out) throws Exception {
        CommandEncoder encoder = ctx.pipeline().get(CommandEncoder.class);
        for (CommandData<?, ?> commandData : msg.getCommands()) {
			// 对单条指令进行编码
            encoder.encode(ctx, commandData, out);
        }
    }