上一篇介绍了Aerosike的基本概念，本文将介绍如何用java连接Aerospike。
一、导入jar包
java程序多用maven构建，Aerospike的pom依赖如下：
<dependencies>
<dependency>
<groupId>com.aerospike</groupId>
<artifactId>aerospike-client</artifactId>
<version>4.0.6</version>
</dependency>
</dependencies>
Aerospike天然支持Netty，如果需要用到Netty做NIO的事件循环，需要增加如下pom依赖：
<dependencies>
<dependency>
<groupId>com.aerospike</groupId>
<artifactId>aerospike-client</artifactId>
<version>4.0.6</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-handler</artifactId>
<version>4.1.11.Final</version>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport</artifactId>
<version>4.1.11.Final</version>
</dependency>
<!-- Only needed when using epoll event loops on linux -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
<classifier>linux-x86_64</classifier>
<version>4.1.11.Final</version>
</dependency>
</dependencies>
部分spark程序用sbt打包，build.sbt内容如下：
libraryDependencies += "com.aerospike" % "aerospike-client" % "latest.integration"
二、单节点连接
根据ip连接客户端，并设置写入策略。
AerospikeClient client = new AerospikeClient("192.168.151.171", 3000);
WritePolicy policy = new WritePolicy();
policy.timeoutDelay = 50;
增
单个key的写入
Key key = new Key("test_namespace","test_setname","test_key");
Bin binString = new Bin("name","value");
client.put(policy,key,binString);
- Key的构造函数对象的三个参数为namespace，set，key的名称
- Bin的构造函数为Bin的name和value。
其中Bin的name为字符串格式，最大长度为14个字符，需要特别注意。
value支持多种格式，入integer、string、double、list、map等，具体支持类型与字段限制，详见：
data type，Known Limitations - client.put函数采用了可变长参数，对应一个key可以存入多个Bins。（源码见附）
删
- Bin的删除
Bin bin = Bin.asNull("name");
client.put(policy,key,bin);
- key的删除
Key key = new Key("test_namespace","test_setname","test_key");
client.delete(policy, key);
改
Aerospike默认写入为覆盖，需要修改Bin时，只要在相同Bin name下重新写入即可。
查
- 根据key查询所有Bins：
Key key = new Key("test_namespace","test_setname","test_key");
Record record = client.get(policy, key);
- 根据key查询指定Bins：
Key key = new Key("test_namespace","test_setname","test_key");
Record record = client.get(policy, key,"bin","bin1");
- Aerospike 同样支持简单的scan，java 需要实现ScanCallback接口
import com.aerospike.client.AerospikeClient;
import com.aerospike.client.AerospikeException;
import com.aerospike.client.Key;
import com.aerospike.client.Record;
import com.aerospike.client.ScanCallback;
import com.aerospike.client.policy.Priority;
import com.aerospike.client.policy.ScanPolicy;
public final class scan implements ScanCallback {
public static void main(String[] args) {
try {
new scan().runTest();
}
catch (Exception e) {
e.printStackTrace();
}
}
private int recordCount;
public void runTest() throws AerospikeException {
AerospikeClient client = new AerospikeClient("192.168.151.171", 3000);
try {
ScanPolicy policy = new ScanPolicy();
policy.concurrentNodes = true;
policy.priority = Priority.LOW;
policy.includeBinData = false;
client.scanAll(policy, "test_namespace","test_setname", this);
System.out.println("Records " + recordCount);
}
finally {
client.close();
}
}
public void scanCallback(Key key, Record record) {
recordCount++;
if ((recordCount % 10000) == 0) {
System.out.println("Records " + recordCount);
}
}
}
其中ScanPolicy 为scan策略的设定，本文设定了三个
-
policy.concurrentNodes = true表示并行的对每一个node执行scan -
policy.priority = Priority.LOW表示优先级为低 -
policy.includeBinData = false表示是否只返回指定Bins,本文设置为否
还有一点需要特别注意的是scanCallback为并行执行，需要保证方法是线程安全的。
三、集群连接
Aerospike集群的数据会自动balance，如果你按照单机连接的方式访问集群，数据会自动分布到整个集群中，但整个集群的性能未全部利用。下面介绍访问集群的方式：
AsyncClientPolicy asyncPolicy = new AsyncClientPolicy();
WritePolicy policy = new WritePolicy();
policy.timeoutDelay = 50;
Host[] hosts = new Host[] {
new Host("192.168.151.171",3000),
new Host("192.168.151.172",3000),
new Host("192.168.151.173",3000),
new Host("192.168.151.174",3000)
};
AsyncClient client = new AsyncClient(asyncPolicy,hosts);
Key key = new Key("test_namespace","test_setname","test_key");
Bin binString = new Bin("cates", "values");
client.put(policy, new NoopWriteListener(),key, binString);
其中NoopWriteListener为自定义类,可以实现成功和失败对应的方法
class NoopWriteListener implements WriteListener {
@Override
public void onSuccess(Key key) {
}
@Override
public void onFailure(AerospikeException exception) {
}
}
其他的增删改查，集群与单机没有明显区别，就不一一介绍了。
四、总结
本文简单介绍了使用java连接Aerospike的方法，更深入的用法如UDF（用户自定义方法），日志处理等问题，有机会再向大家介绍。
附：put函数源码
public final void put(WritePolicy policy, Key key, Bin... bins) throws AerospikeException {
if(policy == null) {
policy = this.writePolicyDefault;
}
WriteCommand command = new WriteCommand(policy, key, bins, Type.WRITE);
command.execute(this.cluster, policy, key, (Node)null, false);
}
