com.basho.riak.client.operations

Class MultiFetchObject<T>

java.lang.Object

com.basho.riak.client.operations.MultiFetchObject<T>

All Implemented Interfaces:

RiakOperation<List<MultiFetchFuture<T>>>

public class MultiFetchObject<T>
extends Object
implements RiakOperation<List<MultiFetchFuture<T>>>

An operation to fetch multiple values from Riak

Use the Bucket.multiFetch(java.lang.String[]), Bucket.multiFetch(java.util.List, java.lang.Class), or Bucket.multiFetch(java.util.List, java.lang.Class) methods to create a mutli-fetch operation.

Riak itself does not support pipelining of requests. The MutliFetchObject addresses this issue by using a threadpool to parallelize a set of fetch operations for a given set of keys.

The result of calling execute() is a List of MultiFetchFuture objects each one representing a fetch operation. The simplest use would be a loop where you iterate through and wait for them to complete:

 List<MultiFetchFuture<MyPojo>> futures = bucket.multiFetch(keys).execute();
 List<MyPojo> myResults = new ArrayList<MyPojo>();
 for (MultiFetchFuture<MyPojo> f : futures)
 {
     try 
     {
          MyPojo mp = f.get();
          myResults.add(mp);
     }
     catch (ExecutionException e)
     {
         // log error, etc.
     }
 }
 
 

Thread Pool:
The internal ThreadPoolExecutor is static; all multi-fetch operations performed by a single instance of the client use the same pool. This is to prevent resource starvation in the case of multiple simultaneous multi-fetch operations. Idle threads (including core threads) are timed out after 5 seconds.

The defaults for corePoolSize is determined by the Java Runtime using:

Runtime.getRuntime().availableProcessors() * 2;

Advanced users can tune this via the setCorePoolSize(int) method; this is passed directly to its counterpart in the ThreadPoolExecutor. The queue feeding the threadpool is unbounded therefore ThreadPoolExecutor.setMaximumPoolSize(int) has no effect and is simply set to match.

Be aware that because requests are being parallelized performance is also dependent on the client's underlying connection pool. If there are no connections available performance will suffer initially as connections will need to be established or worse they could time out.

Author:

Brian Roach

See Also:

Bucket, RiakFactory

Field Summary

Fields

Modifier and Type	Field and Description
static int	DEFAULT_POOL_MAX_SIZE The initial value for both corePoolSize and maximumPoolSize.

Constructor Summary

Constructors

Constructor and Description
MultiFetchObject(RawClient client, String bucket, List<String> keys, Retrier retrier) Create a new MultiFetchOperation that delegates to the given client to fetch the data from bucket for keys using retrier

Method Summary

Methods

Modifier and Type	Method and Description
MultiFetchObject<T>	basicQuorum(boolean basicQuorum)
List<MultiFetchFuture<T>>	execute() Attempts to fetch the data for all the keys, convert it with Converter and resolve any siblings with ConflictResolver
static int	getCorePoolSize() Returns the core number of threads from the internal ThreadPoolExecutor
MultiFetchObject<T>	headOnly() Causes the client to retrieve only the metadata and not the value of this object.
MultiFetchObject<T>	modifiedSince(Date modifiedSince) *NOTE* HTTP Only.
MultiFetchObject<T>	notFoundOK(boolean notFoundOK)
MultiFetchObject<T>	pr(int pr)
MultiFetchObject<T>	pr(Quora pr)
MultiFetchObject<T>	pr(Quorum pr)
MultiFetchObject<T>	r(int r) The read quorum for this fetch operation
MultiFetchObject<T>	r(Quora r) The read quorum for this fetch operation
MultiFetchObject<T>	r(Quorum r) The read quorum for this fetch operation
MultiFetchObject<T>	returnDeletedVClock(boolean returnDeletedVClock)
static void	setCorePoolSize(int size) Sets the core number of threads in the internal ThreadPoolExecutor.
MultiFetchObject<T>	timeout(int timeout) Set an operation timeout in milliseconds to be sent to Riak As of 1.4 Riak allows a timeout to be sent for get, put, and delete operations.
MultiFetchObject<T>	withConverter(Converter<T> converter) A Converter to use to convert the data fetched to some other type
MultiFetchObject<T>	withResolver(ConflictResolver<T> resolver) Sets the ConflictResolver to use for this multi-fetch operation.
MultiFetchObject<T>	withRetrier(Retrier retrier) A Retrier to use

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Detail

DEFAULT_POOL_MAX_SIZE

public static final int DEFAULT_POOL_MAX_SIZE

The initial value for both corePoolSize and maximumPoolSize. This is determined via: Runtime.getRuntime().availableProcessors() * 2;

See Also:

ThreadPoolExecutor

Constructor Detail

MultiFetchObject

public MultiFetchObject(RawClient client,
                String bucket,
                List<String> keys,
                Retrier retrier)

Create a new MultiFetchOperation that delegates to the given client to fetch the data from bucket for keys using retrier

Use Bucket to create a Fetch operation.

Parameters:

client - the RawClient to use for the operation

bucket - the name of the bucket to get the data item from

keys - the keys to get the data items from

retrier - the Retrier to use when executing the operation.

Method Detail

execute

public List<MultiFetchFuture<T>> execute()

Attempts to fetch the data for all the keys, convert it with Converter and resolve any siblings with ConflictResolver

Specified by:

execute in interface RiakOperation<List<MultiFetchFuture<T>>>

Returns:

a List of MultiFetchFuture objects.

setCorePoolSize

public static void setCorePoolSize(int size)

Sets the core number of threads in the internal ThreadPoolExecutor.

Parameters:

size - - the new core size

See Also:

ThreadPoolExecutor.setCorePoolSize(int)

getCorePoolSize

public static int getCorePoolSize()

Returns the core number of threads from the internal ThreadPoolExecutor

Returns:

the core number of threads

See Also:

ThreadPoolExecutor.getCorePoolSize()

withResolver

public MultiFetchObject<T> withResolver(ConflictResolver<T> resolver)

Sets the ConflictResolver to use for this multi-fetch operation.

Parameters:

resolver -

Returns:

this

r

public MultiFetchObject<T> r(int r)

The read quorum for this fetch operation

Parameters:

r - an Integer for the read quorum

Returns:

this

r

public MultiFetchObject<T> r(Quora r)

The read quorum for this fetch operation

Parameters:

r - an Quora for the read quorum

Returns:

this

r

public MultiFetchObject<T> r(Quorum r)

The read quorum for this fetch operation

Parameters:

r - an Quorum for the read quorum

Returns:

this

pr

public MultiFetchObject<T> pr(int pr)

Parameters:

pr -

Returns:

this

See Also:

FetchMeta.Builder.pr(int)

pr

public MultiFetchObject<T> pr(Quora pr)

Parameters:

pr -

Returns:

this

See Also:

FetchMeta.Builder.pr(Quora)

pr

public MultiFetchObject<T> pr(Quorum pr)

Parameters:

pr -

Returns:

this

See Also:

FetchMeta.Builder.pr(Quora)

notFoundOK

public MultiFetchObject<T> notFoundOK(boolean notFoundOK)

Parameters:

notFoundOK -

Returns:

this

See Also:

FetchMeta.Builder.notFoundOK(boolean)

basicQuorum

public MultiFetchObject<T> basicQuorum(boolean basicQuorum)

Parameters:

basicQuorum -

Returns:

this

See Also:

FetchMeta.Builder.basicQuorum(boolean)

timeout

public MultiFetchObject<T> timeout(int timeout)

Set an operation timeout in milliseconds to be sent to Riak As of 1.4 Riak allows a timeout to be sent for get, put, and delete operations. The client will receive a timeout error if the operation is not completed within the specified time

Parameters:

timeout - the timeout in milliseconds

Returns:

this

returnDeletedVClock

public MultiFetchObject<T> returnDeletedVClock(boolean returnDeletedVClock)

Parameters:

returnDeletedVClock -

Returns:

this

See Also:

FetchMeta.Builder.returnDeletedVClock(boolean)

modifiedSince

public MultiFetchObject<T> modifiedSince(Date modifiedSince)

*NOTE* HTTP Only.

Parameters:

modifiedSince - a last modified date.

Returns:

this

headOnly

public MultiFetchObject<T> headOnly()

Causes the client to retrieve only the metadata and not the value of this object. Note if you are using HTTP: If siblings are present the client does a second get and retrieves all the values. This is due to how the HTTP API handles siblings. Note: The Converter being used must be able to handle an empty value.

Returns:

this

See Also:

FetchMeta.Builder.headOnly(boolean headOnly)

withConverter

public MultiFetchObject<T> withConverter(Converter<T> converter)

A Converter to use to convert the data fetched to some other type

Parameters:

converter -

Returns:

this

withRetrier

public MultiFetchObject<T> withRetrier(Retrier retrier)

A Retrier to use

Parameters:

retrier -

Returns:

this