c# - Cassandra 并行插入性能 c#

Question

在单个节点 i7 8cores 上测试 Cassandra C# 插入性能时，插入速度仅为 100 次/秒。使用 Datastax Cassandra 驱动程序是否可以改进代码？尝试了异步和同步 Session.Execute，但性能很差。

[TestMethod]
public void TestMethod2()
{
    //   CREATE TABLE table1(
    //   col1 text,
    //   col2 timestamp,
    //   col3 int,
    //   col4 text,
    //   PRIMARY KEY(col1, col2, col3)
    //   );

    PoolingOptions poolingOptions = new PoolingOptions();
    poolingOptions
        .SetCoreConnectionsPerHost(HostDistance.Local, 1280)
        .SetMaxConnectionsPerHost(HostDistance.Local, 1280)
        .SetCoreConnectionsPerHost(HostDistance.Remote, 1280)
        .SetMaxConnectionsPerHost(HostDistance.Remote, 1280);

    poolingOptions
        .SetMaxSimultaneousRequestsPerConnectionTreshold(HostDistance.Local, 32768)
        .SetMinSimultaneousRequestsPerConnectionTreshold(HostDistance.Remote, 2000);

    var cluster = Cluster.Builder()
        .AddContactPoints("localhost")
        .WithPoolingOptions(poolingOptions)
        .WithQueryOptions(new QueryOptions().SetConsistencyLevel(ConsistencyLevel.One))
        .WithLoadBalancingPolicy(new RoundRobinPolicy())
        .Build();

    var options = new ParallelOptions();
    options.MaxDegreeOfParallelism = 50;

    using (var session = cluster.Connect("keyspace1"))
    {
        var ps = session.Prepare("INSERT INTO table1(col1,col2,col3,col4) VALUES (?,?,?,?)");

        var r = Parallel.For(0, 1000, options, (x) =>
        {
            {
                var statement = ps.Bind("123456", DateTime.UtcNow, x, "1234 some log message goes here. Hello world. 123334556567586978089-==00");
                var t = session.ExecuteAsync(statement);
                t.Wait();
            }
        });
    }
}

Answer 1

连接数量 (> 1K) 太大了，对于现代 Cassandra 实例 (2.1+)，单个连接就足够了。

执行异步操作的最快方法是启动 n 个操作，每次任务完成时启动一个新任务。

使用 SemaphoreSlim 可以限制并发操作的数量，或者您可以使用 continuations 手动完成。

如果您不想编写该代码，可以使用我创建的一个小实用程序包 ConcurrentUtils :

// Execute 1,000,000 times
// limiting the maximum amount of parallel async operations to 512
await ConcurrentUtils.Times(1000000, 512, _ => 
  session.ExecuteAsync(ps.Bind(parameters)));