上一篇介绍了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策略的设定，本文设定了三个

1. policy.concurrentNodes = true表示并行的对每一个node执行scan
2. policy.priority = Priority.LOW表示优先级为低
3. 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);
    }