c++ - 使用 : "Operation timed out" 进行大量插入后，cassandra INSERT 失败

Question

我使用 cassandra-c++-driver 在 100 列的表中写入 100000 行，如下所示:

#include <cstdlib>
#include <stdio.h>
#include <cassandra.h>
#include <string>
#include <iostream>
#include <random>
#include <chrono>
#include <unistd.h>
#include <thread>
CassFuture *connect_future = NULL;
CassCluster *cluster = NULL;
CassSession *session = NULL;
std::random_device rd;
std::mt19937_64 gen(rd());
std::uniform_int_distribution<unsigned long long> dis;


int COLUMNS_COUNT = 100;
using namespace std;

void insertQ() {

    auto t1 = std::chrono::high_resolution_clock::now();


    for (int row = 0; row < 10000; ++row) {
        string columns;
        for (int i = 0; i < COLUMNS_COUNT; ++i) {
            columns += "name" + to_string(i) + " , ";
        }

        string result = "INSERT INTO mykeyspace.users2 (user_id,";
        result += columns;
        result += "lname) VALUES (";


        string values = to_string(dis(gen) % 50000000) + ",";

        for (int i = 0; i < COLUMNS_COUNT; ++i) {
            values += "'name" + to_string(dis(gen)) + "' , ";
        }

        values += " 'lname" + to_string(dis(gen) % 20) + "'";
        result += values;
        result += ");";


        CassStatement *statement = cass_statement_new(result.c_str(), 0);

        CassFuture *result_future = cass_session_execute(session, statement);
        cass_future_wait(result_future);

        if (cass_future_error_code(result_future) == CASS_OK) {
//            cout << "insert ok" << endl;
        }
        else {
            const char *message;
            size_t message_length;
            cass_future_error_message(result_future, &message, &message_length);
            fprintf(stderr, "Unable to run query: '%.*s'\n", (int) message_length,
                    message);

            cerr << "index : " << row << endl;
        }

        cass_statement_free(statement);
        cass_future_free(result_future);

        if (row % 1000 == 0)
        {
//            usleep(1000000);
//            std::this_thread::sleep_for(std::chrono::seconds(1));
//            cass_se
        }

    }
    auto t2 = std::chrono::high_resolution_clock::now();

    auto duration = std::chrono::duration_cast<std::chrono::milliseconds>(t2 - t1);

    cout << "duration: " << duration.count() << endl;


}


int main() {

    /* Setup and connect to cluster */
    connect_future = NULL;
    cluster = cass_cluster_new();
    session = cass_session_new();

    /* Add contact points */
//    cass_cluster_set_contact_points(cluster, "127.0.0.1,127.0.0.2,127.0.0.3");
    cass_cluster_set_contact_points(cluster, "127.0.0.1");

    /* Provide the cluster object as configuration to connect the session */
    connect_future = cass_session_connect(session, cluster);

    if (cass_future_error_code(connect_future) == CASS_OK) {
        CassFuture *close_future = NULL;

        insertQ();

        /* Close the session */
        close_future = cass_session_close(session);
        cass_future_wait(close_future);
        cass_future_free(close_future);
    } else {
        /* Handle error */
        const char *message;
        size_t message_length;
        cass_future_error_message(connect_future, &message, &message_length);
        fprintf(stderr, "Unable to connect: '%.*s'\n", (int) message_length,
                message);
    }

    cass_future_free(connect_future);
    cass_cluster_free(cluster);
    cass_session_free(session);

    return 0;
}

它工作并写入了大约 90000 行，然后出现此错误:
索引:91627
无法运行查询:“操作超时 - 仅收到 0 个响应。”
..

然后继续，我可以执行“SELECT”查询，但在此“INSERT”失败之后。 unitl 我重新启动 cassandra 服务。

有什么问题吗？
我的系统:Ubuntu 14.04 x64、8 gig ram、cassandra 2.1.4(来自具有默认配置的 cassandra debian 存储库)

谢谢。

Answer 1

此错误从 Cassandra 返回。它表示在 cassandra 中配置的时间段内响应您的读/写请求的拷贝数少于所需的数量。由于您没有指定一致性级别，因此所需要的只是一个节点响应并且它不在写入超时范围内。 cassandra.yaml 中最相关的配置是:

write_request_timeout_in_ms (default 2000ms)
read_request_timeout_in_ms (default: 5000ms)
range_request_timeout_in_ms (default: 10000ms)

由于您正在进行插入，write_request_timeout_in_ms 可能是最相关的配置。

可能发生的情况是您的 cassandra 集群不堪重负。在运行测试时，您是否查看了服务器上的 cpu 利用率/磁盘 io/内存利用率？

有趣的是，您的代码一次只执行 1 个 INSERT，这是正确的吗？我希望这应该没问题，但也许正在发生的事情是这对你在 cassandra 中的内存堆施加了巨大的压力并且它不能足够快地刷新数据，所以它在写入磁盘时会被备份。您应该查看您的 cassandra system.log(如果在 Linux 上，通常在/var/log/cassandra 中)并查看是否有任何关于长垃圾回收(查找 GCInspector)或内存表压力的可疑消息。