ClickHouse R2dbc 使用整理

1.pom

        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-r2dbc</artifactId>
        </dependency>
        <dependency>
            <groupId>com.clickhouse</groupId>
            <artifactId>clickhouse-r2dbc</artifactId>
            <version>0.4.6</version>
            <classifier>all</classifier>
            <exclusions>
                <exclusion>
                    <groupId>*</groupId>
                    <artifactId>*</artifactId>
                </exclusion>
            </exclusions>
        </dependency>

2.启动类

@SpringBootApplication(exclude = {R2dbcAutoConfiguration.class, R2dbcDataAutoConfiguration.class})
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

3.配置类

import io.r2dbc.spi.ConnectionFactories;
import io.r2dbc.spi.ConnectionFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ClickhouseConfig {

    @Bean
    public ConnectionFactory getClickhouseConnect(){
        //r2dbc:clickhouse:http://{username}:{password}@{host}:{port}/{database}
        return ConnectionFactories.get(String.format("r2dbc:clickhouse:http://%s:%s@%s:%d/%s",{username},{password},{host},{port},{database}));
    }

}

4.ck 操作

参考官方例子，亲测，可用

@Repository
public class ClickRepository {

    @Autowired
    ConnectionFactory connectionFactory;

    public Flux<ClickStats> getStatsByDomain(String domain){
        return Mono.from(connectionFactory.create())
                .flatMapMany(conn -> conn.createStatement("select domain, path,  toDate(cdate) as d, count(1) as count from clickdb.clicks where domain = :domain group by domain, path, d")
                        .bind("domain", domain)
                        .execute())
                .flatMap(result -> result.map((row, rowMetadata) -> new ClickStats(row
                        .get("domain", String.class), row.get("path", String.class), row.get("d", LocalDate.class),  row.get("count", Long.class))));
    }

    public Mono<Void> add(Click click){
        return Mono.from(connectionFactory.create())
                .flatMapMany(conn -> execute(click, conn)).then();
    }

    private Publisher<? extends Result> execute(Click click, Connection conn) {
        return conn.createStatement("insert into clickdb.clicks values (:domain, :path, :cdate, :count)")
        .bind("domain", click.getDomain())
        .bind("path", click.getPath())
                .bind("cdate", LocalDateTime.now())
                .bind("count", 1).execute();
    }

}

上面的配置是连接单节点，集群配置

参考源码：

ClickHouseNodes 中会解析url中host判断是否有“,”分隔符。

所以集群配置：

//r2dbc:clickhouse:http://{username}:{password}@{host}:{port}/{database}

host 可以配置多个，直接用“,”分割，亲测，可用。