Java 类com.datastax.driver.core.QueryLogger 实例源码

/**
 * Creates a session and ensures schema if configured. Closes the cluster and session if any
 * exception occurred.
 */
@Override public Session create(Cassandra3Storage cassandra) {
  Closer closer = Closer.create();
  try {
    Cluster cluster = closer.register(buildCluster(cassandra));
    cluster.register(new QueryLogger.Builder().build());
    Session session;
    if (cassandra.ensureSchema) {
      session = closer.register(cluster.connect());
      Schema.ensureExists(cassandra.keyspace, session);
      session.execute("USE " + cassandra.keyspace);
    } else {
      session = cluster.connect(cassandra.keyspace);
    }

    initializeUDTs(session);

    return session;
  } catch (RuntimeException e) {
    try {
      closer.close();
    } catch (IOException ignored) {
    }
    throw e;
  }
}

/**
 * Creates a session and ensures schema if configured. Closes the cluster and session if any
 * exception occurred.
 */
@Override public Session create(CassandraStorage cassandra) {
  Closer closer = Closer.create();
  try {
    Cluster cluster = closer.register(buildCluster(cassandra));
    cluster.register(new QueryLogger.Builder().build());
    if (cassandra.ensureSchema) {
      Session session = closer.register(cluster.connect());
      Schema.ensureExists(cassandra.keyspace, session);
      session.execute("USE " + cassandra.keyspace);
      return session;
    } else {
      return cluster.connect(cassandra.keyspace);
    }
  } catch (RuntimeException e) {
    try {
      closer.close();
    } catch (IOException ignored) {
    }
    throw e;
  }
}

public CassandraStorage(ReaperApplicationConfiguration config, Environment environment) {
  CassandraFactory cassandraFactory = config.getCassandraFactory();
  overrideQueryOptions(cassandraFactory);
  overrideRetryPolicy(cassandraFactory);
  overridePoolingOptions(cassandraFactory);
  cassandra = cassandraFactory.build(environment);
  if (config.getActivateQueryLogger()) {
    cassandra.register(QueryLogger.builder().build());
  }
  CodecRegistry codecRegistry = cassandra.getConfiguration().getCodecRegistry();
  codecRegistry.register(new DateTimeCodec());
  session = cassandra.connect(config.getCassandraFactory().getKeyspace());

  initializeAndUpgradeSchema(cassandra, session, config.getCassandraFactory().getKeyspace());
  prepareStatements();
}

private Cluster doCreateCluster(CassandraProperties properties) {
    Cluster cluster = Cluster.builder()
            .withClusterName(properties.getCluster())
            .withPort(properties.getPort())
            .addContactPoints(properties.getContactPoints())
            .withTimestampGenerator(getTimestampGenerator())
            .withPoolingOptions(
                    //TODO some default options - move to config
                    new PoolingOptions()
                            .setConnectionsPerHost(HostDistance.LOCAL, 4, 4)
                            .setConnectionsPerHost(HostDistance.REMOTE, 2, 2)
                            .setMaxRequestsPerConnection(HostDistance.LOCAL, 1024)
                            .setMaxRequestsPerConnection(HostDistance.REMOTE, 256)
            )
            .build();
    //almost all queries are idempotent except counter updates, so it's easier to mark them as idempotent
    cluster.getConfiguration().getQueryOptions().setDefaultIdempotence(true);

    CodecRegistry codecRegistry = cluster.getConfiguration().getCodecRegistry();

    TupleType tupleType = cluster.getMetadata()
            .newTupleType(DataType.timestamp(), DataType.varchar());
    codecRegistry.register(new ZonedDateTimeCodec(tupleType));

    QueryLogger queryLogger = QueryLogger.builder()
            .withConstantThreshold(100)
            .withMaxQueryStringLength(200)
            .build();
    cluster.register(queryLogger);

    return cluster;
}

/**
 * Method to create the standalone cassandra connection .
 * 
 * @param ip
 * @param port
 * @param userName
 * @param password
 * @param keyspace
 * @return
 */
private boolean createStandaloneConnection(String ip, String port, String userName,
    String password, String keyspace) {

  Session cassandraSession = null;
  boolean connection = false;
  Cluster cluster = null;
  try {
    if (null == cassandraSessionMap.get(keyspace)) {
      PropertiesCache cache = PropertiesCache.getInstance();
      PoolingOptions poolingOptions = new PoolingOptions();
      poolingOptions.setCoreConnectionsPerHost(HostDistance.LOCAL,
          Integer.parseInt(cache.getProperty(Constants.CORE_CONNECTIONS_PER_HOST_FOR_LOCAL)));
      poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL,
          Integer.parseInt(cache.getProperty(Constants.MAX_CONNECTIONS_PER_HOST_FOR_LOCAl)));
      poolingOptions.setCoreConnectionsPerHost(HostDistance.REMOTE,
          Integer.parseInt(cache.getProperty(Constants.CORE_CONNECTIONS_PER_HOST_FOR_REMOTE)));
      poolingOptions.setMaxConnectionsPerHost(HostDistance.REMOTE,
          Integer.parseInt(cache.getProperty(Constants.MAX_CONNECTIONS_PER_HOST_FOR_REMOTE)));
      poolingOptions.setMaxRequestsPerConnection(HostDistance.LOCAL,
          Integer.parseInt(cache.getProperty(Constants.MAX_REQUEST_PER_CONNECTION)));
      poolingOptions.setHeartbeatIntervalSeconds(
          Integer.parseInt(cache.getProperty(Constants.HEARTBEAT_INTERVAL)));
      poolingOptions
          .setPoolTimeoutMillis(Integer.parseInt(cache.getProperty(Constants.POOL_TIMEOUT)));
      if (!ProjectUtil.isStringNullOREmpty(userName)
          && !ProjectUtil.isStringNullOREmpty(password)) {
        cluster = createCluster(ip, port, userName, password, poolingOptions);
      } else {
        cluster = createCluster(ip, port, poolingOptions);
      }
      QueryLogger queryLogger = QueryLogger.builder().withConstantThreshold(
          Integer.parseInt(cache.getProperty(Constants.QUERY_LOGGER_THRESHOLD))).build();
      cluster.register(queryLogger);
      cassandraSession = cluster.connect(keyspace);

      if (null != cassandraSession) {
        connection = true;
        cassandraSessionMap.put(keyspace, cassandraSession);
        cassandraclusterMap.put(keyspace, cluster);
      }
      final Metadata metadata = cluster.getMetadata();
      String msg = String.format("Connected to cluster: %s", metadata.getClusterName());
      ProjectLogger.log(msg);

      for (final Host host : metadata.getAllHosts()) {
        msg = String.format("Datacenter: %s; Host: %s; Rack: %s", host.getDatacenter(),
            host.getAddress(), host.getRack());
        ProjectLogger.log(msg);
      }
    }
  } catch (Exception e) {
    ProjectLogger.log("Error occured while creating connection :", e);
    throw new ProjectCommonException(ResponseCode.internalError.getErrorCode(), e.getMessage(),
        ResponseCode.SERVER_ERROR.getResponseCode());
  }

  if (null != cassandraSessionMap.get(keyspace)) {
    connection = true;
  }
  return connection;


}