1. 序章
SpringアプリケーションでのCouchbaseの使用に関するチュートリアルのフォローアップでは、Couchbase SDKの非同期性と、Couchbase SDKを使用して永続化操作をバッチで実行する方法を探り、アプリケーションでCouchbaseリソースの最適な使用。
1.1. CrudServiceインターフェース
まず、汎用の CrudService インターフェースを拡張して、バッチ操作を含めます。
public interface CrudService<T> {
...
List<T> readBulk(Iterable<String> ids);
void createBulk(Iterable<T> items);
void updateBulk(Iterable<T> items);
void deleteBulk(Iterable<String> ids);
boolean exists(String id);
}
1.2. CouchbaseEntityインターフェース
永続化するエンティティのインターフェイスを定義します。
public interface CouchbaseEntity {
String getId();
void setId(String id);
}
1.3. AbstractCrudServiceクラス
次に、これらの各メソッドを汎用抽象クラスに実装します。 このクラスは、前のチュートリアルで使用した PersonCrudService クラスから派生し、次のように始まります。
public abstract class AbstractCrudService<T extends CouchbaseEntity> implements CrudService<T> {
private BucketService bucketService;
private Bucket bucket;
private JsonDocumentConverter<T> converter;
public AbstractCrudService(BucketService bucketService, JsonDocumentConverter<T> converter) {
this.bucketService = bucketService;
this.converter = converter;
}
protected void loadBucket() {
bucket = bucketService.getBucket();
}
...
}
2. 非同期バケットインターフェイス
Couchbase SDKは、非同期操作を実行するためのAsyncBucketインターフェイスを提供します。 Bucket インスタンスを指定すると、 async()メソッドを介して非同期バージョンを取得できます。
AsyncBucket asyncBucket = bucket.async();
3. バッチ操作
AsyncBucket インターフェースを使用してバッチ操作を実行するには、RxJavaライブラリを使用します。
3.1. バッチ読み取り
ここでは、readBulkメソッドを実装します。 まず、 AsyncBucket そしてその flatMap ドキュメントを非同期で取得するRxJavaのメカニズム観察可能 、次に使用します toBlocking のメカニズム RxJava これらをエンティティのリストに変換するには:
@Override
public List<T> readBulk(Iterable<String> ids) {
AsyncBucket asyncBucket = bucket.async();
Observable<JsonDocument> asyncOperation = Observable
.from(ids)
.flatMap(new Func1<String, Observable<JsonDocument>>() {
public Observable<JsonDocument> call(String key) {
return asyncBucket.get(key);
}
});
List<T> items = new ArrayList<T>();
try {
asyncOperation.toBlocking()
.forEach(new Action1<JsonDocument>() {
public void call(JsonDocument doc) {
T item = converter.fromDocument(doc);
items.add(item);
}
});
} catch (Exception e) {
logger.error("Error during bulk get", e);
}
return items;
}
3.2. バッチ挿入
ここでも、RxJavaのflatMap構造を使用して、createBulkメソッドを実装します。
バルクミューテーションリクエストは、応答を生成するよりも速く生成され、オーバーロード状態になることがあるため、 BackpressureException が発生するたびに、指数関数的な遅延で再試行を開始します。
@Override
public void createBulk(Iterable<T> items) {
AsyncBucket asyncBucket = bucket.async();
Observable
.from(items)
.flatMap(new Func1<T, Observable<JsonDocument>>() {
@SuppressWarnings("unchecked")
@Override
public Observable<JsonDocument> call(final T t) {
if(t.getId() == null) {
t.setId(UUID.randomUUID().toString());
}
JsonDocument doc = converter.toDocument(t);
return asyncBucket.insert(doc)
.retryWhen(RetryBuilder
.anyOf(BackpressureException.class)
.delay(Delay.exponential(TimeUnit.MILLISECONDS, 100))
.max(10)
.build());
}
})
.last()
.toBlocking()
.single();
}
3.3. バッチ更新
updateBulkメソッドでも同様のメカニズムを使用します。
@Override
public void updateBulk(Iterable<T> items) {
AsyncBucket asyncBucket = bucket.async();
Observable
.from(items)
.flatMap(new Func1<T, Observable<JsonDocument>>() {
@SuppressWarnings("unchecked")
@Override
public Observable<JsonDocument> call(final T t) {
JsonDocument doc = converter.toDocument(t);
return asyncBucket.upsert(doc)
.retryWhen(RetryBuilder
.anyOf(BackpressureException.class)
.delay(Delay.exponential(TimeUnit.MILLISECONDS, 100))
.max(10)
.build());
}
})
.last()
.toBlocking()
.single();
}
3.4. バッチ削除
そして、deleteBulkメソッドを次のように記述します。
@Override
public void deleteBulk(Iterable<String> ids) {
AsyncBucket asyncBucket = bucket.async();
Observable
.from(ids)
.flatMap(new Func1<String, Observable<JsonDocument>>() {
@SuppressWarnings("unchecked")
@Override
public Observable<JsonDocument> call(String key) {
return asyncBucket.remove(key)
.retryWhen(RetryBuilder
.anyOf(BackpressureException.class)
.delay(Delay.exponential(TimeUnit.MILLISECONDS, 100))
.max(10)
.build());
}
})
.last()
.toBlocking()
.single();
}
4. PersonCrudService
最後に、PersonエンティティのAbstractCrudServiceを拡張するSpringサービスPersonCrudServiceを記述します。
すべてのCouchbaseインタラクションは抽象クラスで実装されているため、エンティティクラスの実装は簡単です。これは、すべての依存関係が挿入され、バケットがロードされていることを確認するだけでよいためです。
@Service
public class PersonCrudService extends AbstractCrudService<Person> {
@Autowired
public PersonCrudService(
@Qualifier("TutorialBucketService") BucketService bucketService,
PersonDocumentConverter converter) {
super(bucketService, converter);
}
@PostConstruct
private void init() {
loadBucket();
}
}
5. 結論
このチュートリアルに示されているソースコードは、githubプロジェクトで入手できます。
Couchbase Java SDKの詳細については、公式のCouchbase開発者向けドキュメントサイトを参照してください。