Java连接cassandra实现简单的增删查demo

package com.draven.merchant.utils; import com.datastax.driver.core.Cluster; import com.datastax.driver.core.ConsistencyLevel; import com.datastax.driver.core.ProtocolOptions; import com.datastax.driver.core.QueryOptions; import com.datastax.driver.core.policies.DCAwareRoundRobinPolicy; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.InitializingBean; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; /** * cassandra cluster 工厂接口 * Created by Draven on 2017/6/13. * Email:hjx@qiandw.com */ @Service("clusterFactoryBean") public class ClusterFactoryBeanImpl implements ClusterFactoryBean, InitializingBean, DisposableBean { private Cluster cluster = null; /** * 接触点, 逗号分隔 */ @Value("${db.cassandra.contactPoint}") private String contactNodes; /** * 端口，0为使用默认值 */ private int port = 0; /** * 用户名, 空为不验证 */ @Value("${db.cassandra.username}") private String username; /** * 密码 */ @Value("${db.cassandra.password}") private String password; /** * 本地数据中心，空则使用默认负载均衡策略 */ private String datacenter; /** * 是否压缩，是则使用LZ4压缩数据 */ @Value("${db.cassandra.compression}") private Boolean compression = false; /** * 创建cluster链接 */ private synchronized void createCluster() throws Exception { cluster = getBuilder().build(); cluster.init(); } @Override public Cluster getCluster() throws Exception { if (cluster == null || cluster.isClosed()) { createCluster(); } return cluster; } @Override public void destroy() throws Exception { cluster.close(); } @Override public void afterPropertiesSet() throws Exception { createCluster(); } private Cluster.Builder getBuilder() throws Exception { Cluster.Builder b = Cluster.builder(); if (StringUtils.isEmpty(contactNodes)) { throw new Exception("No contact point defined."); } String[] cps = contactNodes.split(","); for (String cp : cps) { b.addContactPoint(cp.trim()); } if (port > 0) { b.withPort(port); } if (StringUtils.isNotEmpty(username) && StringUtils.isNotEmpty(password)) { b.withCredentials(username, password); } if (StringUtils.isNotEmpty(datacenter)) { b.withLoadBalancingPolicy(DCAwareRoundRobinPolicy.builder().withLocalDc(datacenter).build()); } if (compression) { b.withCompression(ProtocolOptions.Compression.LZ4); } //设置默认的语句属性 QueryOptions qos = new QueryOptions(); qos.setConsistencyLevel(ConsistencyLevel.LOCAL_QUORUM); b.withQueryOptions(qos); return b; } public void setCluster(Cluster cluster) { this.cluster = cluster; } public String getContactNodes() { return contactNodes; } public void setContactNodes(String contactNodes) { this.contactNodes = contactNodes; } public int getPort() { return port; } public void setPort(int port) { this.port = port; } public String getUsername() { return username; } public void setUsername(String username) { this.username = username; } public String getPassword() { return password; } public void setPassword(String password) { this.password = password; } public String getDatacenter() { return datacenter; } public void setDatacenter(String datacenter) { this.datacenter = datacenter; } public Boolean getCompression() { return compression; } public void setCompression(Boolean compression) { this.compression = compression; } }