Java 类com.datastax.driver.core.policies.RoundRobinPolicy 实例源码

public CassandraConfigDb(List<String> contactPoints, int port) {

    this.contactPoints = new ArrayList<InetAddress> (contactPoints.size());

    for (String contactPoint : contactPoints) {
        try {
            this.contactPoints.add(InetAddress.getByName(contactPoint));
        } catch (UnknownHostException e) {
               throw new IllegalArgumentException(e.getMessage());
        }
    }

    this.port = port;

    cluster = (new Cluster.Builder()).withPort (this.port)
            .addContactPoints(this.contactPoints)
            .withSocketOptions(new SocketOptions().setReadTimeoutMillis(60000).setKeepAlive(true).setReuseAddress(true))
            .withLoadBalancingPolicy(new RoundRobinPolicy())
            .withReconnectionPolicy(new ConstantReconnectionPolicy(500L))
            .withQueryOptions(new QueryOptions().setConsistencyLevel(ConsistencyLevel.ONE))
            .build ();

    session = cluster.newSession();
    preparedStatements = new ConcurrentHashMap<StatementName, PreparedStatement> ();
    prepareStatementCreateLock = new Object();
}

static Cluster buildCluster(Cassandra3Storage cassandra) {
  Cluster.Builder builder = Cluster.builder();
  List<InetSocketAddress> contactPoints = parseContactPoints(cassandra);
  int defaultPort = findConnectPort(contactPoints);
  builder.addContactPointsWithPorts(contactPoints);
  builder.withPort(defaultPort); // This ends up protocolOptions.port
  if (cassandra.username != null && cassandra.password != null) {
    builder.withCredentials(cassandra.username, cassandra.password);
  }
  builder.withRetryPolicy(ZipkinRetryPolicy.INSTANCE);
  builder.withLoadBalancingPolicy(new TokenAwarePolicy(new LatencyAwarePolicy.Builder(
      cassandra.localDc != null
          ? DCAwareRoundRobinPolicy.builder().withLocalDc(cassandra.localDc).build()
          : new RoundRobinPolicy()
      // This can select remote, but LatencyAwarePolicy will prefer local
  ).build()));
  builder.withPoolingOptions(new PoolingOptions().setMaxConnectionsPerHost(
      HostDistance.LOCAL, cassandra.maxConnections
  ));
  return builder.build();
}

static Cluster buildCluster(CassandraStorage cassandra) {
  Cluster.Builder builder = Cluster.builder();
  List<InetSocketAddress> contactPoints = parseContactPoints(cassandra);
  int defaultPort = findConnectPort(contactPoints);
  builder.addContactPointsWithPorts(contactPoints);
  builder.withPort(defaultPort); // This ends up protocolOptions.port
  if (cassandra.username != null && cassandra.password != null) {
    builder.withCredentials(cassandra.username, cassandra.password);
  }
  builder.withRetryPolicy(ZipkinRetryPolicy.INSTANCE);
  builder.withLoadBalancingPolicy(new TokenAwarePolicy(new LatencyAwarePolicy.Builder(
      cassandra.localDc != null
          ? DCAwareRoundRobinPolicy.builder().withLocalDc(cassandra.localDc).build()
          : new RoundRobinPolicy()
      // This can select remote, but LatencyAwarePolicy will prefer local
  ).build()));
  builder.withPoolingOptions(new PoolingOptions().setMaxConnectionsPerHost(
      HostDistance.LOCAL, cassandra.maxConnections
  ));
  return builder.build();
}

/**
 * Get a Cassandra cluster using hosts and port.
 */
private Cluster getCluster(List<String> hosts, int port, String username, String password,
                           String localDc, String consistencyLevel) {
  Cluster.Builder builder = Cluster.builder()
      .addContactPoints(hosts.toArray(new String[0]))
      .withPort(port);

  if (username != null) {
    builder.withAuthProvider(new PlainTextAuthProvider(username, password));
  }

  if (localDc != null) {
    builder.withLoadBalancingPolicy(
        new TokenAwarePolicy(new DCAwareRoundRobinPolicy.Builder().withLocalDc(localDc).build()));
  } else {
    builder.withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()));
  }

  if (consistencyLevel != null) {
    builder.withQueryOptions(
        new QueryOptions().setConsistencyLevel(ConsistencyLevel.valueOf(consistencyLevel)));
  }

  return builder.build();
}

/**
 * Connect to a cassandra cluster at a given host/port
 */
public void connect() {
    try {
        lock.lock();
    } catch (IOException e) {
        throw new IllegalStateException("There appears to be another health check running", e);
    }
    final List<InetSocketAddress> whiteList= new ArrayList<>();
    whiteList.add(new InetSocketAddress(host, port));

    final LoadBalancingPolicy loadBalancingPolicy = new WhiteListPolicy(new RoundRobinPolicy(), whiteList);
    final Cluster.Builder cb = Cluster.builder()
            .addContactPoint(host)
            .withPort(port)
            .withLoadBalancingPolicy(loadBalancingPolicy)
            .withRetryPolicy(retryPolicy);
    if (username != null) {
        cb.withCredentials(username, password);
    }
    cluster = cb.build();
    session = cluster.connect();
    hosts = cluster.getMetadata().getAllHosts();
}

public Cluster cluster() {

        if (cluster != null) return cluster;
        String[] entryPoints = System.getProperty("cassandra.servers", "localhost").split(",");
        String clusterName = System.getProperty("cassandra.cluster-name", "Test Cluster");
        int port = Integer.getInteger("cassandra.port", 9042);
        log.info("Connecting the cluster {} via hosts {} with port {}", clusterName, Arrays.toString(entryPoints), port);
        Cluster.Builder builder = Cluster.builder()
                .addContactPoints(entryPoints)
                .withClusterName(clusterName)
                .withPort(port)
                .withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()));
        cluster = builder.build();
        return cluster;

    }

@Test
public void cassandraSessionWithConfiguration() throws Exception {
    ApplicationContext testContext = getTestApplicationContext(
            "cloud-cassandra-with-config.xml", createService("my-service"));
    Cluster cluster = testContext.getBean("cassandra-full-config",
            getConnectorType());

    assertNotNull(cluster.getConfiguration().getSocketOptions());
    assertEquals(15000,
            cluster.getConfiguration().getSocketOptions().getConnectTimeoutMillis());
    assertTrue(DefaultRetryPolicy.class.isAssignableFrom(
            cluster.getConfiguration().getPolicies().getRetryPolicy().getClass()));
    assertTrue(RoundRobinPolicy.class.isAssignableFrom(cluster.getConfiguration()
            .getPolicies().getLoadBalancingPolicy().getClass()));
    assertTrue(ConstantReconnectionPolicy.class.isAssignableFrom(cluster
            .getConfiguration().getPolicies().getReconnectionPolicy().getClass()));
}

/**
 * Currently we connect just once and then reuse the connection.
 * We do not bother with closing the connection.
 *
 * It is normal to use one Session per DB. The Session is thread safe.
 */
private void connect() {

    if (cluster == null) {

        log.info("Connecting to Cassandra server on " + this.dbHost + " at port " + this.dbPort);

        // allow fetching as much data as present in the DB
        QueryOptions queryOptions = new QueryOptions();
        queryOptions.setFetchSize(Integer.MAX_VALUE);
        queryOptions.setConsistencyLevel(ConsistencyLevel.ONE);

        cluster = Cluster.builder()
                         .addContactPoint(this.dbHost)
                         .withPort(this.dbPort)
                         .withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()))
                         .withReconnectionPolicy(new ExponentialReconnectionPolicy(500, 30000))
                         .withQueryOptions(queryOptions)
                         .withCredentials(this.dbUser, this.dbPassword)
                         .build();

    }

    if (session == null) {

        log.info("Connecting to Cassandra DB with name " + this.dbName);
        session = cluster.connect(dbName);
    }
}

@Bean("cassandraDataSource")
@Primary
public DataSource createDataSource() {
    DataSource dataSource = new DataSource();
    dataSource.setContactPoints("127.0.0.1");
    dataSource.setPort(9042);
    dataSource.setReadConsistency(ConsistencyLevel.ONE.name());
    dataSource.setWriteConsistency(ConsistencyLevel.ONE.name());
    dataSource.setLoadBalancingPolicy(new RoundRobinPolicy());
    return dataSource;
}

@Test
public void loadBalancing_defaultsToRoundRobin() {
  RoundRobinPolicy policy = toRoundRobinPolicy(Cassandra3Storage.builder().build());

  Host foo = mock(Host.class);
  when(foo.getDatacenter()).thenReturn("foo");
  Host bar = mock(Host.class);
  when(bar.getDatacenter()).thenReturn("bar");
  policy.init(mock(Cluster.class), asList(foo, bar));

  assertThat(policy.distance(foo)).isEqualTo(HostDistance.LOCAL);
  assertThat(policy.distance(bar)).isEqualTo(HostDistance.LOCAL);
}

RoundRobinPolicy toRoundRobinPolicy(Cassandra3Storage storage) {
  return (RoundRobinPolicy) ((LatencyAwarePolicy) ((TokenAwarePolicy) buildCluster(storage)
      .getConfiguration()
      .getPolicies()
      .getLoadBalancingPolicy())
      .getChildPolicy()).getChildPolicy();
}

@Test
public void loadBalancing_defaultsToRoundRobin() {
  RoundRobinPolicy policy = toRoundRobinPolicy(CassandraStorage.builder().build());

  Host foo = mock(Host.class);
  when(foo.getDatacenter()).thenReturn("foo");
  Host bar = mock(Host.class);
  when(bar.getDatacenter()).thenReturn("bar");
  policy.init(mock(Cluster.class), asList(foo, bar));

  assertThat(policy.distance(foo)).isEqualTo(HostDistance.LOCAL);
  assertThat(policy.distance(bar)).isEqualTo(HostDistance.LOCAL);
}

RoundRobinPolicy toRoundRobinPolicy(CassandraStorage storage) {
  return (RoundRobinPolicy) ((LatencyAwarePolicy) ((TokenAwarePolicy) buildCluster(storage)
      .getConfiguration()
      .getPolicies()
      .getLoadBalancingPolicy())
      .getChildPolicy()).getChildPolicy();
}

public AsyncFuture<Connection> construct() {
    AsyncFuture<Session> session = async.call(() -> {
        // @formatter:off
        final PoolingOptions pooling = new PoolingOptions();

        final QueryOptions queryOptions = new QueryOptions()
            .setFetchSize(fetchSize)
            .setConsistencyLevel(consistencyLevel);

        final SocketOptions socketOptions = new SocketOptions()
            .setReadTimeoutMillis((int) readTimeout.toMilliseconds());

        final Cluster.Builder cluster = Cluster.builder()
            .addContactPointsWithPorts(seeds)
            .withRetryPolicy(retryPolicy)
            .withPoolingOptions(pooling)
            .withQueryOptions(queryOptions)
            .withSocketOptions(socketOptions)
            .withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()));
        // @formatter:on

        authentication.accept(cluster);
        return cluster.build().connect();
    });

    if (configure) {
        session = session.lazyTransform(s -> {
            return schema.configure(s).directTransform(i -> s);
        });
    }

    return session.lazyTransform(s -> {
        return schema.instance(s).directTransform(schema -> {
            return new Connection(s, schema);
        });
    });
}

@Test
public void testLoadBalancingPolicyParsing() throws Exception
{
    String lbPolicyStr = "RoundRobinPolicy()";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof RoundRobinPolicy);
    System.out.println("====================");
    lbPolicyStr = "TokenAwarePolicy(RoundRobinPolicy())";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof TokenAwarePolicy);
    System.out.println("====================");
    lbPolicyStr = "DCAwareRoundRobinPolicy(\"dc1\")";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof DCAwareRoundRobinPolicy);
    System.out.println("====================");
    lbPolicyStr = "TokenAwarePolicy(DCAwareRoundRobinPolicy(\"dc1\"))";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof TokenAwarePolicy);       
    System.out.println("====================");
    lbPolicyStr = "TokenAwarePolicy";
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr)==null);
    System.out.println("====================");
    lbPolicyStr = "LatencyAwarePolicy(TokenAwarePolicy(RoundRobinPolicy()),(double) 10.5,(long) 1,(long) 10,(long)1,10)";       
    System.out.println(lbPolicyStr);
    assertTrue(Utils.parseLbPolicy(lbPolicyStr) instanceof LatencyAwarePolicy);
    System.out.println("====================");             

}

public MetaStoreClient(String... contactPoints) {
    if (contactPoints.length == 0)
        throw new RuntimeException("No contact points specified");

    cluster = Cluster.builder()
            .addContactPoints(contactPoints)
            .withClusterName(Schema.CLUSTER)
            .withLoadBalancingPolicy(new RoundRobinPolicy())
            .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
            .withReconnectionPolicy(new ExponentialReconnectionPolicy(100, 10000))
            .withoutMetrics()
            .build();
}

private Cluster getNewCluster(String cassandraNodes) {
    return Cluster.builder()
            .withoutJMXReporting()
            .withoutMetrics()
            .addContactPoints(cassandraNodes.split(","))
            .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
            .withReconnectionPolicy(new ExponentialReconnectionPolicy(100L, TimeUnit.MINUTES.toMillis(5)))
            .withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()))
            .build();
}

@Test
public void buildsPolicy() throws Exception {
    final RoundRobinPolicyFactory factory = new RoundRobinPolicyFactory();

    final LoadBalancingPolicy policy = factory.build();

    assertThat(policy).isExactlyInstanceOf(RoundRobinPolicy.class);
}

@Provides @Singleton Cluster provideCluster() {
    Cluster cluster = Cluster.builder()
            .addContactPoints("127.0.0.1")
            .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
            .withReconnectionPolicy(new ExponentialReconnectionPolicy(100L, 5000L))
            .withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy()))
            .build();
    Metadata metadata = cluster.getMetadata();
    LOGGER.info("Connected to cluster: '{}'", metadata.getClusterName());
    metadata.getAllHosts()
            .forEach(host -> LOGGER.info("Datacenter: '{}'; Host: '{}'; Rack: '{}'",
                    new Object[] { host.getDatacenter(), host.getAddress(), host.getRack() })
            );
    return cluster;
}

private void displayClusterInfo() {
    Metadata metadata = m_cluster.getMetadata();
    m_logger.info("Connected to cluster with topography:");
    RoundRobinPolicy policy = new RoundRobinPolicy();
    for (Host host : metadata.getAllHosts()) {
        m_logger.info("   Host {}: datacenter: {}, rack: {}, distance: {}",
                      new Object[]{host.getAddress(), host.getDatacenter(), 
            host.getRack(), policy.distance(host)});
    }
    m_logger.info("Database contains {} keyspaces", metadata.getKeyspaces().size());
}

@Test
public void shouldCreateClusterWithConfig() throws Exception {

    CassandraServiceInfo info = new CassandraServiceInfo("local",
            Collections.singletonList("127.0.0.1"), 9142);

    CassandraClusterConfig config = new CassandraClusterConfig();
    config.setCompression(ProtocolOptions.Compression.NONE);
    config.setPoolingOptions(new PoolingOptions().setPoolTimeoutMillis(1234));
    config.setQueryOptions(new QueryOptions());
    config.setProtocolVersion(ProtocolVersion.NEWEST_SUPPORTED);
    config.setLoadBalancingPolicy(new RoundRobinPolicy());
    config.setReconnectionPolicy(new ConstantReconnectionPolicy(1));
    config.setRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE);
    config.setSocketOptions(new SocketOptions());

    Cluster cluster = creator.create(info, config);

    Configuration configuration = cluster.getConfiguration();

    assertThat(configuration.getProtocolOptions().getCompression(),
            is(config.getCompression()));
    assertThat(configuration.getQueryOptions(), is(config.getQueryOptions()));
    assertThat(configuration.getSocketOptions(), is(config.getSocketOptions()));

    Policies policies = configuration.getPolicies();
    assertThat(policies.getLoadBalancingPolicy(),
            is(config.getLoadBalancingPolicy()));
    assertThat(policies.getReconnectionPolicy(), is(config.getReconnectionPolicy()));
    assertThat(policies.getRetryPolicy(), is(config.getRetryPolicy()));
}

@Provides
@Named("pooledmetacluster")
Cluster providePooledCluster(@Named("staash.cassclient") String clientType,@Named("staash.metacluster") String clustername) {
    if (clientType.equals("cql")) {
    Cluster cluster = Cluster.builder().withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy())).addContactPoint(clustername).build();
    return cluster;
    }else {
        return null;
    }
}

@Provides
    @Named("pooledmetacluster")
    Cluster providePooledCluster(@Named("paas.cassclient") String clientType,@Named("paas.metacluster") String clustername) {
        if (clientType.equals("cql")) {
        Cluster cluster = Cluster.builder().withLoadBalancingPolicy(new TokenAwarePolicy(new RoundRobinPolicy())).addContactPoint(clustername).build();
//        Cluster cluster = Cluster.builder().addContactPoint(clustername).build();
        return cluster;
        }else {
            return null;
        }
    }

@PostConstruct
public void initialize() {
    String cassandraHosts = env.getProperty("ea.cassandra.hosts","localhost:9042");
    String cassandraClusterName = env.getProperty("ea.cassandra.cluster","ElasticActorsCluster");
    String cassandraKeyspaceName = env.getProperty("ea.cassandra.keyspace","\"ElasticActors\"");
    Integer cassandraPort = env.getProperty("ea.cassandra.port", Integer.class, 9042);

    Set<String> hostSet = StringUtils.commaDelimitedListToSet(cassandraHosts);

    String[] contactPoints = new String[hostSet.size()];
    int i=0;
    for (String host : hostSet) {
        if(host.contains(":")) {
            contactPoints[i] = host.substring(0,host.indexOf(":"));
        } else {
            contactPoints[i] = host;
        }
        i+=1;
    }

    PoolingOptions poolingOptions = new PoolingOptions();
    poolingOptions.setHeartbeatIntervalSeconds(60);
    poolingOptions.setConnectionsPerHost(HostDistance.LOCAL, 2, env.getProperty("ea.cassandra.maxActive",Integer.class,Runtime.getRuntime().availableProcessors() * 3));
    poolingOptions.setPoolTimeoutMillis(2000);

    Cluster cassandraCluster =
            Cluster.builder().withClusterName(cassandraClusterName)
                    .addContactPoints(contactPoints)
                    .withPort(cassandraPort)
            .withLoadBalancingPolicy(new RoundRobinPolicy())
            .withRetryPolicy(new LoggingRetryPolicy(DefaultRetryPolicy.INSTANCE))
            .withPoolingOptions(poolingOptions)
            .withReconnectionPolicy(new ConstantReconnectionPolicy(env.getProperty("ea.cassandra.retryDownedHostsDelayInSeconds",Integer.class,1) * 1000))
            .withQueryOptions(new QueryOptions().setConsistencyLevel(ConsistencyLevel.QUORUM)).build();

    this.cassandraSession = cassandraCluster.connect(cassandraKeyspaceName);


}

private static Cluster build(List<String> addresses, int port) {
  Cluster.Builder builder = Cluster.builder();
  for (String address : addresses) {
    builder.addContactPoint(address);
  }
  builder
    .withPort(port)
    .withCompression(Compression.LZ4)
    .withRetryPolicy(DowngradingConsistencyRetryPolicy.INSTANCE)
    .withLoadBalancingPolicy(LatencyAwarePolicy.builder(new RoundRobinPolicy()).build());

  Cluster cluster = builder.build();

  try {
    // Attempt to init the cluster to make sure it's usable. I'd prefer to remove this and leave it on the
    // client to retry when the connect method throws an exception.
    cluster.init();
    return cluster;
  } catch(NoHostAvailableException e) {
    LOGGER.warn("Unable to connect to Cassandra, will retry contact points next time",
        cluster, e);
    cluster = builder.build();
    cluster.init();
  }

  return cluster;

}

public static LoadBalancingPolicy getLoadBalancingPolicy() {

        LoadBalancingPolicy loadBalancingPolicy = new TokenAwarePolicy(new RoundRobinPolicy());
        return loadBalancingPolicy;
    }

public WhiteListPolicyWithOnePriorityNode(/*LoadBalancingPolicy childPolicy, */Host primaryHost, Collection<Host> allNodesWithReplica, Collection<InetSocketAddress> hostAddress) {
        super(new RoundRobinPolicy(), hostAddress);
        this.primaryHost = primaryHost;
//        allNodesWithReplica.remove(primaryHost); // no need to do this.
        this.remainingNodes = allNodesWithReplica;
    }

@Override
public LoadBalancingPolicy build() {
    return new RoundRobinPolicy();
}

private synchronized void initClient()
   {
       // We should be able to go without an Atomic variable here.  There are two potential problems:
       // 1. Multiple threads read intialized=false and call init client.  However, the method is
       //    synchronized so only one will get the lock first, and the others will drop out here.
       // 2. One thread reads initialized=true before initClient finishes.  This also should not
       //    happen as the lock should include a memory barrier.
       if (initialized || initializationFailed)
           return;

    // Just while we initialise the client, we must temporarily
    // disable all logging or else we get into an infinite loop
    Level globalThreshold = LogManager.getLoggerRepository().getThreshold();
    LogManager.getLoggerRepository().setThreshold(Level.OFF);

    try
       {
           Cluster.Builder builder = Cluster.builder()
                                            .addContactPoints(hosts.split(",\\s*"))
                                            .withPort(port)
                                            .withLoadBalancingPolicy(new RoundRobinPolicy());

           // Kerberos provides authentication anyway, so a username and password are superfluous.  SSL
           // is compatible with either.
           boolean passwordAuthentication = !password.equals("") || !username.equals("");
           if (authProviderOptions != null && passwordAuthentication)
               throw new IllegalArgumentException("Authentication via both Cassandra usernames and Kerberos " +
                                                  "requested.");

           // Encryption
           if (authProviderOptions != null)
               builder = builder.withAuthProvider(getAuthProvider());
           if (sslOptions != null)
               builder = builder.withSSL(getSslOptions());
           if (passwordAuthentication)
               builder = builder.withCredentials(username, password);

           cluster = builder.build();
        session = cluster.connect();
           setupSchema();
           setupStatement();
    }
       catch (Exception e)
       {
        LogLog.error("Error ", e);
        errorHandler.error("Error setting up cassandra logging schema: " + e);

           //If the user misconfigures the port or something, don't keep failing.
           initializationFailed = true;
    }
       finally
       {
           //Always reenable logging
           LogManager.getLoggerRepository().setThreshold(globalThreshold);
           initialized = true;
    }
}