import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
/**
 * Bean post init method. The following configuration is used
 * for initializing the ThresholdsDao cassandra cluster,
 * <p>
 * <ul>
 * <li>Active clients per node - 4</li>
 * <li>Cassandra Thrift timeout - 600 sec</li>
 * </ul>
 */
public void init() {
    logger.info("Initializing Monitor Thresholds Dao...");

    Cluster cluster = cb.getCluster(clusterName, 4, timeout);
    logger.info("Connected to cluster : " + clusterName);

    SchemaBuilder.createSchema(cluster, keyspaceName);
    ConfigurableConsistencyLevel cl = new ConfigurableConsistencyLevel();
    cl.setDefaultWriteConsistencyLevel(HConsistencyLevel.ONE);
    cl.setDefaultReadConsistencyLevel(HConsistencyLevel.ONE);

    keyspace = HFactory.createKeyspace(keyspaceName, cluster, cl);
    thresholdMutator = HFactory.createMutator(keyspace, longSerializer);
    manifestMapMutator = HFactory.createMutator(keyspace, longSerializer);
    realizedAsMutator = HFactory.createMutator(keyspace, longSerializer);
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
@Test
public void testCassandraConsistencyPolicy() {
	Configuration configuration = ConfigUtils.getInstance().getConfiguration();

	// The default configuration should be to set the consistency policies
	// to Read.ONE and Write.ONE
	HectorConsistencyPolicy policy = new HectorConsistencyPolicy("NoCF");
	assertEquals(HConsistencyLevel.ONE, policy.get(OperationType.READ));
	assertEquals(HConsistencyLevel.ONE, policy.get(OperationType.WRITE));

	// Now set explicit policies and test those.

	// Start with read properties.

	checkOperationConfigurationLoading(configuration,
									   HecubaConstants.GLOBAL_PROP_NAME_PREFIX + ".consistencypolicy.read",
									   OperationType.READ);
	checkOperationConfigurationLoading(configuration,
									   HecubaConstants.GLOBAL_PROP_NAME_PREFIX + ".consistencypolicy.write",
									   OperationType.WRITE);

}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
@Override
public void prepare(Map stormConf, TopologyContext context,
		OutputCollector collector) {
	ouc = collector;
	// load from properties file
	Properties prop = new Properties();
	try {
		prop.load(new FileInputStream("config.properties"));
	} catch (IOException ex) {
		log.error(ex.toString());
	}
	
	//cassandra configuration
	cluster = HFactory.getOrCreateCluster(
			prop.getProperty("CLUSTERNAME"), prop.getProperty("HOST"));
	ConfigurableConsistencyLevel ccl = new ConfigurableConsistencyLevel();
	ccl.setDefaultReadConsistencyLevel(HConsistencyLevel.ONE);
	keyspace = HFactory.createKeyspace("links", cluster, ccl,
			FailoverPolicy.FAIL_FAST);
	mutator = HFactory.createMutator(keyspace, StringSerializer.get());
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
@Override public Keyspace getOrCreateKeyspace(String keyspaceName, int replicationFactor, Cluster cluster) {
    KeyspaceDefinition keyspaceDef = cluster.describeKeyspace(keyspaceName);
    List<ColumnFamilyDefinition> cfDefs = new LinkedList<ColumnFamilyDefinition>();
    // If keyspace does not exist create it without CFs.
    if (keyspaceDef == null) {
        keyspaceDef = HFactory.createKeyspaceDefinition(keyspaceName, ThriftKsDef.DEF_STRATEGY_CLASS, replicationFactor, cfDefs);
        cluster.addKeyspace(keyspaceDef);
    }

    Keyspace keyspace = HFactory.createKeyspace(keyspaceName, cluster);

    ConfigurableConsistencyLevel consistencyLevel = new ConfigurableConsistencyLevel();
    consistencyLevel.setDefaultReadConsistencyLevel(HConsistencyLevel.QUORUM);
    consistencyLevel.setDefaultWriteConsistencyLevel(HConsistencyLevel.QUORUM);
    keyspace.setConsistencyLevelPolicy(consistencyLevel);

    return keyspace;
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
/**
 * Bean post init method. The following configuration is used
 * for initializing the OpsEventDao cassandra cluster,
 * <p>
 * <ul>
 * <li>Active clients per node - 4</li>
 * <li>Cassandra Thrift timeout - 5 sec </li>
 * </ul>
 */
public void init() {
    logger.info("Initializing OpsEvent Dao...");
    Cluster cluster = cb.getCluster(clusterName, 4, 5 * 1000);
    logger.info("Connected to cluster : " + clusterName);

    SchemaBuilder.createSchema(cluster, keyspaceName);
    ConfigurableConsistencyLevel cl = new ConfigurableConsistencyLevel();
    cl.setDefaultWriteConsistencyLevel(HConsistencyLevel.ONE);
    cl.setDefaultReadConsistencyLevel(HConsistencyLevel.ONE);
    keyspace = createKeyspace(keyspaceName, cluster, cl);
    eventMutator = HFactory.createMutator(keyspace, bytesSerializer);
    ciMutator = HFactory.createMutator(keyspace, longSerializer);
    orphanCloseMutator = HFactory.createMutator(keyspace, longSerializer);
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
/**
 * Bean post init method. The following configuration is used
 * for initializing the OpsCiStateDao cassandra cluster,
 * <p/>
 * <ul>
 * <li>Active clients per node - 4</li>
 * <li>Cassandra Thrift timeout - 5 sec </li>
 * </ul>
 */
public void init() {
    logger.info("Initializing OpsCiState Dao...");
    Cluster cluster = cb.getCluster(clusterName, 4, 5 * 1000);
    logger.info("Connected to cluster : " + clusterName);

    SchemaBuilder.createSchema(cluster, keyspaceName);
    ConfigurableConsistencyLevel cl = new ConfigurableConsistencyLevel();
    HConsistencyLevel wrCL = System.getProperty("com.sensor.cistates.cl","LOCAL_QUORUM").equals("ONE") ? HConsistencyLevel.ONE :HConsistencyLevel.LOCAL_QUORUM;
    cl.setDefaultWriteConsistencyLevel(wrCL);
    cl.setDefaultReadConsistencyLevel(HConsistencyLevel.ONE);
    keyspace = createKeyspace(keyspaceName, cluster, cl);
    ciStateHistMutator = HFactory.createMutator(keyspace, longSerializer);
    componentStateMutator = HFactory.createMutator(keyspace, longSerializer);
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
/**
 * Method in charge of setting the consistency level for defined column families.
 * @param pColFams  Column families
 * @return Map<String, HConsistencyLevel> with the mapping between colFams and consistency level.
 */
private Map<String, HConsistencyLevel> getConsisLevelForColFams(List<ColumnFamilyDefinition> pColFams) {
  Map<String, HConsistencyLevel> clMap = new HashMap<>();
  // Get columnFamily consistency level.
  String colFamConsisLvl = (colFamConsLvl != null && !colFamConsLvl.isEmpty())?colFamConsLvl:DEFAULT_HECTOR_CONSIS_LEVEL;
  LOG.debug("ColumnFamily consistency level configured to '" + colFamConsisLvl + "'.");
  // Define consistency for ColumnFamily "ColumnFamily"
  for (ColumnFamilyDefinition colFamDef : pColFams)
    clMap.put(colFamDef.getName(), HConsistencyLevel.valueOf(colFamConsisLvl));
  return clMap;
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
private HConsistencyLevel configureConsistencyLevel(String columnFamilyName, String operation) {
	// first get the default consistency policy.
	String[] consistencyPolicyProperties = HecubaConstants.getConsistencyPolicyProperties(columnFamilyName, operation);
	String consistencyParameter = "ONE";
	for (String property : consistencyPolicyProperties) {
		consistencyParameter = ConfigUtils.getInstance().getConfiguration().getString(property, consistencyParameter);
	}

	return StringUtils.isEmpty(consistencyParameter) ? HConsistencyLevel.ONE : HConsistencyLevel.valueOf(consistencyParameter);
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
@Override
public HConsistencyLevel get(OperationType operationType) {
	switch (operationType) {
	case READ:
		return readConsistency;
	case WRITE:
		return writeConsistency;
	default:
		return HConsistencyLevel.ONE; // Just in Case
	}
}
 

import me.prettyprint.hector.api.HConsistencyLevel; //导入依赖的package包/类
private void checkOperationConfigurationLoading(Configuration configuration, String operationProperty,
												OperationType operationType) {
	HectorConsistencyPolicy policy;
	for (HConsistencyLevel consistencyLevel : HConsistencyLevel.values()) {
		configuration.setProperty(operationProperty, consistencyLevel.toString());
		policy = new HectorConsistencyPolicy("NoCF");
		assertEquals("Error handling consistency level " + consistencyLevel.toString() +
							 " within HectorConsistencyPolicy.", consistencyLevel, policy.get(operationType));
	}
}