Cassandra入门试用

1. 安装

1.1 官方文档

http://cassandra.apache.org/doc/latest/getting_started/index.html

1.2 安装前提

• 安装Java 8

$java -version
java version "1.8.0_102"
Java(TM) SE Runtime Environment (build 1.8.0_102-b14)

• 安装Python 2.7

$python -V
Python 2.7.17

1.3 安装cassandra

• 下载

wget http://mirrors.tuna.tsinghua.edu.cn/apache/cassandra/3.11.5/apache-cassandra-3.11.5-bin.tar.gz

• 解压即安装

tar -xzvf apache-cassandra-3.11.5-bin.tar.gz

• 启动脚本在cassandra的根目录的bin下

./bin/cassandra -f

2. 集群配置

• 案例: 配置了3节点集群

  • 位于tdatanode1、tdatanode2、tdatanode3
  • tdatanode1、tdatanode2做为种子节点
  • 3个安装过程同上

• 配置文件修改, $CASSANDRA_HOME/conf/cassandra.yaml

  • 所有机器使用相同的集群名称

      cluster_name: 'utrackRealTime'
  

  • 所有机器设置seed，均为tdatanode1、tdatanode2

      seed_provider:
          - class_name: org.apache.cassandra.locator.SimpleSeedProvider
            parameters:
                - seeds: "tdatanode1,tdatanode2"
  

  • 监听地址设置为本机IP或者host名称

      listen_address: tdatanode1
      rpc_address: tdatanode1
  

• 启动集群

# -f的意思的前台运行，测试时为了更方便的查看问题
./bin/cassandra -f

• 交互式终端cqlsh连接

# 如果不是连本机，通过环境变量$CQLSH_HOST、$CQLSH_PORT指定主机和端口
cqlsh --request-timeout=600
# 指定机器
cqlsh tdatanode1 --request-timeout=60000

3. Java客户端

3.1 Maven依赖

<dependency>
    <groupId>com.datastax.cassandra</groupId>
    <artifactId>cassandra-driver-core</artifactId>
    <version>3.1.0</version>
</dependency>

3.2 客户端代码

• 创建Session，应该是线程安全的，可以重用

PoolingOptions options = new PoolingOptions();
options.setNewConnectionThreshold(HostDistance.LOCAL,1800);
options.setCoreConnectionsPerHost(HostDistance.LOCAL,5);
options.setMaxConnectionsPerHost(HostDistance.LOCAL,5);

Cluster.Builder bulder = Cluster.builder().withPoolingOptions(options).addContactPoint("192.168.36.174").withPort(9042);
Cluster cluster = builder.build();
Session session = cluster.connect();

• 执行操作

ResultSet rs = session.execute("select * FROM utrack.odl_user_hj");
for (Row r : rs.all()) {
    int size = r.getColumnDefinitions().size();
    for (int i = 0; i < size; i++) {
        r.getObject(i).toString();
    }
}

• 进程退出时的清理

session.close();
cluster.close();

4. 性能测试

4.1 压测结论

连接	并发	QPS
1	1	1452
1	5	8182
1	10	14806
1	20	23584
1	30	32341
1	40	35945
1	50	33647

单连接、3节点、普通HDD、单JVM分配8G内存的情况下，并发30~40基本到最高性能， 35K QPS

连接数	并发数	QPS	备注
1	40	35945
L -5	40	36179	- 设置LOCAL最大连接数没有用，估计是
L -5	40
NewConnectionThreshold=0		25994
NewConnectionThreshold=1		29620
NewConnectionThreshold=10		32840
NewConnectionThreshold=800		37481
NotSet		37407
NewConnectionThreshold=1800		37000	- 阈值到800后再增长，对qps没有明显影响
R 5-10	40	35842
L 5-10	40	24113	- LOCAL coreConnect设置为5的情况下，性能反而差了。

不确定系统瓶颈的情况下，不要随意调优，Connection和新建Connection的Threshold调整可能不但不提升性能，反而降低

4.2 压测代码

• 主类

public class ReadPerformance {

    private static ExecutorService es = Executors.newFixedThreadPool(40);

    public static void main(String[] args) throws InterruptedException {
        Session s = CassandraConnectionManager.getSession();
        PreparedStatement preparedStatement = s.prepare("select * from utrack.simple_test where domain = ?");

        long start = System.currentTimeMillis();

        for (int i = 0; i < 10_0000; i++) {
            final int index = i;
            es.submit(() -> {
                ResultSet rs = s.execute(preparedStatement.bind((int) (Math.random() * 200_0000)));
                rs.one().getObject(0);
                if (index % 1000 == 999) System.out.println("index: " + index + ", completed");
            });
        }

        System.out.println("await termination....");

        es.shutdown();

        if (!es.awaitTermination(1, TimeUnit.HOURS)) {
            System.out.println("await timeout....");
            es.shutdownNow();
        }

        long last = System.currentTimeMillis() - start;

        System.out.println("time used: " + last + ", qps:" + (10_0000_000 / last));

        CassandraConnectionManager.destory();

    }
}

• 工具类

public class CassandraConnectionManager {

    public static Session getSession() {
        return Holder.session;
    }

    public static Cluster getCluster() {
        return Holder.cluster;
    }

    public static void destory() {
        Holder.destroy();
    }

    private static class Holder {
        private static Cluster cluster;
        private static Session session;

        static {
            PoolingOptions options = new PoolingOptions();
            options.setNewConnectionThreshold(HostDistance.LOCAL,1800);
//            options.setCoreConnectionsPerHost(HostDistance.LOCAL,5);
            options.setMaxConnectionsPerHost(HostDistance.LOCAL,5);

            Cluster.Builder bulder = Cluster.builder().withPoolingOptions(options).addContactPoint("192.168.36.174").withPort(9042);
            cluster = bulder.build();
            session = cluster.connect();
        }

        private static void destroy() {
            if (session != null) session.close();
            if (cluster != null) cluster.close();
        }
    }

}