如何将文件部分流式传输到数据库（r2dbcpostgres）？

为了将文件部分流式传输到Postgres数据库，我们需要使用r2dbc-postgresql驱动程序，并遵循以下步骤：

1. 准备表格

创建具有bytea（二进制）数据类型的表。 例如，

CREATE TABLE files (
  id serial PRIMARY KEY,
  name text NOT NULL,
  content bytea NOT NULL
);

2. 使用Flux传输文件内容

使用Flux将文件内容流式传输到数据插入查询中。例如，

public Mono<Integer> save(String fileName, Flux<DataBuffer> data){
    return databaseClient.execute("INSERT INTO files(name, content) VALUES ($1, $2)")
       .bind("$1", fileName)
       .bind("$2", data.asByteArray())
       .fetch().rowsUpdated();
}

请注意，我们需要将Flux<DataBuffer>转换为byte[]。可以使用以下代码：

Mono<byte[]> bytes = data.reduce(DataBuffer::write).map(DataBuffer::asByteBuffer).map(ByteBuffer::array);

3. 完整代码示例

import io.netty.buffer.ByteBufAllocator;
import io.r2dbc.postgresql.codec.DefaultCodecs;
import io.r2dbc.postgresql.extension.CodecRegistrar;
import io.r2dbc.postgresql.extension.Extension;
import io.r2dbc.postgresql.extension.ExtensionConfiguration;
import io.r2dbc.postgresql.extension.ExtensionRegistry;
import io.r2dbc.spi.Blob;
import io.r2dbc.spi.Blob.CapacityMismatchException;
import io.r2dbc.spi.Blob.InvalidFormatException;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.ConnectionFactory;
import io.r2dbc.spi.R2dbcNonTransientResourceException;
import io.r2dbc.spi.Result;
import io.r2dbc.spi.Statement;
import io.r2dbc.spi.Wrapped;
import org.reactivestreams.Publisher;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.DataBufferUtils;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.netty.ByteBufFlux;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;

public class ByteaDataBufferCodec implements CodecRegistrar {
    @Override
    public void register(Extension extension) {
        extension.addCodec(ByteDataBuffer.class, new ByteaDataBufferCodecImpl(new DefaultCodecs()));
    }

    @Override
    public String getCodecRegistrarName() {
        return "ByteaDataBufferCodec";
    }

    static final class ByteaDataBufferCodecImpl implements io.r2dbc.postgresql.codec.ByteBufCodec<ByteData