无法使用 Hector 连接到 cassandra

Question

我无法使用 Hector 访问 Casandra。以下是代码

 import java.util.Arrays;
 import java.util.List;
 import me.prettyprint.cassandra.service.CassandraHostConfigurator;
 import me.prettyprint.cassandra.service.ThriftCluster;
 import me.prettyprint.cassandra.service.ThriftKsDef;
 import me.prettyprint.hector.api.Cluster;
 import me.prettyprint.hector.api.Keyspace;
 import me.prettyprint.hector.api.ddl.ColumnFamilyDefinition;
 import me.prettyprint.hector.api.ddl.KeyspaceDefinition;
 import me.prettyprint.hector.api.factory.HFactory;
 import me.prettyprint.hector.api.mutation.Mutator;

 public class Hector {
 public static void main (String[] args){
boolean cfExists = false;
Cluster cluster = HFactory.getOrCreateCluster("mycluster", new                     CassandraHostConfigurator("host:9160"));
Keyspace keyspace = HFactory.createKeyspace("Keyspace1", cluster);
// first check if the key space exists
        KeyspaceDefinition keyspaceDetail = cluster.describeKeyspace("Keyspace1");
        // if not, create one
        if (keyspaceDetail == null) {

            CassandraHostConfigurator cassandraHostConfigurator = new CassandraHostConfigurator("host:9160");
            ThriftCluster cassandraCluster = new ThriftCluster("mycluster", cassandraHostConfigurator);

            ColumnFamilyDefinition cfDef = HFactory.createColumnFamilyDefinition("Keyspace1", "base");
            cassandraCluster.addKeyspace(new ThriftKsDef("Keyspace1", "org.apache.cassandra.locator.SimpleStrategy", 1,
                    Arrays.asList(cfDef)));

        } else {

            // even if the key space exists, we need to check if the column family exists
            List<ColumnFamilyDefinition> columnFamilyDefinitions = keyspaceDetail.getCfDefs();
            for (ColumnFamilyDefinition def : columnFamilyDefinitions)    {
                String columnFamilyName = def.getName();
                if (columnFamilyName.equals("tcs_im"))
                    cfExists = true;
            }
        }
 }
 } 

遇到以下错误

log4j:WARN 找不到记录器的附加程序（me.prettyprint.cassandra.connection.CassandraHostRetryService）。 log4j:WARN 请正确初始化 log4j 系统。 log4j:WARN 请参阅http://logging.apache.org/log4j/1.2/faq.html#noconfig 了解更多信息。 线程“主”java.lang.IllegalAccessError 中的异常：试图从类 me.prettyprint.cassandra.connection.HConnectionManager 访问类 me.prettyprint.cassandra.service.JmxMonitor 在 me.prettyprint.cassandra.connection.HConnectionManager.(HConnectionManager.java:78) 在 me.prettyprint.cassandra.service.AbstractCluster.(AbstractCluster.java:69) 在 me.prettyprint.cassandra.service.AbstractCluster.(AbstractCluster.java:65) 在 me.prettyprint.cassandra.service.ThriftCluster.(ThriftCluster.java:17) 在 me.prettyprint.hector.api.factory.HFactory.createCluster(HFactory.java:176) 在 me.prettyprint.hector.api.factory.HFactory.getOrCreateCluster(HFactory.java:155) 在 com.im.tcs.Hector.main(Hector.java:20)

请帮忙看看为什么会这样。

Answer 1

我们使用 CassandraConnection 类作为便利类：

import me.prettyprint.cassandra.connection.DynamicLoadBalancingPolicy;
import me.prettyprint.cassandra.service.CassandraHostConfigurator;
import me.prettyprint.cassandra.service.ExhaustedPolicy;
import me.prettyprint.cassandra.service.OperationType;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.HConsistencyLevel;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.factory.HFactory;

import java.util.HashMap;
import java.util.Map;

/**
 * lazy connect
 */
final class CassandraConnection {

    // Constants -----------------------------------------------------

    private static final String HOSTS = "localhost";
    private static final int PORT = "9160";
    private static final String CLUSTER_NAME = "myCluster";
    private static final int TIMEOUT = 500);
    private static final String KEYSPACE = "Keyspace1";
    private static final ConsistencyLevelPolicy CL_POLICY = new ConsistencyLevelPolicy();

    // Attributes ----------------------------------------------------

    private Cluster cluster;
    private volatile Keyspace keyspace;

    // Constructors --------------------------------------------------

    CassandraConnection() {}

    // Methods --------------------------------------------------------

    Cluster getCluster() {
        if (null == cluster) {
            CassandraHostConfigurator config = new CassandraHostConfigurator();
            config.setHosts(HOSTS);
            config.setPort(PORT);
            config.setUseThriftFramedTransport(true);
            config.setUseSocketKeepalive(true);
            config.setAutoDiscoverHosts(false);
            // maxWorkerThreads provides the throttling for us. So hector can be let to grow freely...
            config.setExhaustedPolicy(ExhaustedPolicy.WHEN_EXHAUSTED_GROW);
            config.setMaxActive(1000); // hack since ExhaustedPolicy doesn't work
            // suspend hosts if response is unacceptable for web response
            config.setCassandraThriftSocketTimeout(TIMEOUT);
            config.setUseHostTimeoutTracker(true);
            config.setHostTimeoutCounter(3);
            config.setLoadBalancingPolicy(new DynamicLoadBalancingPolicy());

            cluster = HFactory.createCluster(CLUSTER_NAME, config);

        }
        return cluster;
    }

    Keyspace getKeyspace() {
        if (null == keyspace) {
            keyspace = HFactory.createKeyspace(KEYSPACE, getCluster(), CL_POLICY);
        }
        return keyspace;
    }

    private static class ConsistencyLevelPolicy implements me.prettyprint.hector.api.ConsistencyLevelPolicy {

        @Override
        public HConsistencyLevel get(final OperationType op) {
            return HConsistencyLevel.ONE;
        }

        @Override
        public HConsistencyLevel get(final OperationType op, final String cfName) {
            return get(op);
        }
    }
}

使用示例：

private final CassandraConnection conn = new CassandraConnection();

SliceQuery<String, String, String> sliceQuery = HFactory.createSliceQuery(
                conn.getKeyspace(), StringSerializer.get(), StringSerializer.get(), StringSerializer.get());
sliceQuery.setColumnFamily("myColumnFamily");
sliceQuery.setRange("", "", false, Integer.MAX_VALUE);
sliceQuery.setKey("myRowKey");
ColumnSlice<String, String> columnSlice = sliceQuery.execute().get();