Kestrel Component
The Kestrel component allows messages to be sent to a Kestrel queue, or messages to be consumed from a Kestrel queue. This component uses the spymemcached
client for memcached protocol communication with Kestrel servers.
URI format
kestrel://[addresslist/]queuename[?options]
Where queuename is the name of the queue on Kestrel. The addresslist part of the URI may include one or more host:port
pairs. For example, to connect to the queue foo
on kserver01:22133
, use:
kestrel://kserver01:22133/foo
If the addresslist is omitted, localhost:22133
is assumed, i.e.:
kestrel://foo
Likewise, if a port is omitted from a host:port
pair in addresslist, the default port 22133 is assumed, i.e.:
kestrel://kserver01/foo
Here is an example of a Kestrel endpoint URI used for producing to a clustered queue:
kestrel://kserver01:22133,kserver02:22133,kserver03:22133/massive
Here is an example of a Kestrel endpoint URI used for consuming concurrently from a queue:
kestrel://kserver03:22133/massive?concurrentConsumers=25&waitTimeMs=500
Options
You can configure properties on each Kestrel endpoint individually by specifying them in the ?parameters
portion of the endpoint URI. Any ?parameters
that are omitted will default to what is configured on the KestrelComponent's base KestrelConfiguration. The following properties may be set on KestrelConfiguration and/or each individual endpoint:
Option |
Default Value |
Description |
---|---|---|
|
|
Specifies the number of concurrent consumer threads. |
|
|
Specifies the |
NOTE: If waitTimeMs is set to zero (or negative), the /t=...
specifier does not get passed to the server on GET requests. When a queue is empty, the GET call returns immediately with no value. In order to prevent "tight looping" in the polling phase, this component will do a Thread.sleep(100)
whenever nothing is returned from the GET request (only when nothing is returned). You are highly encouraged to configure a positive non-zero value for waitTimeMs.
Configuring the Kestrel component using Spring XML
The simplest form of explicit configuration is as follows:
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd"> <bean id="kestrel" class="org.apache.camel.component.kestrel.KestrelComponent"/> <camelContext xmlns="http://camel.apache.org/schema/spring"> </camelContext> </beans>
That will enable the Kestrel component with all default settings, i.e. it will use localhost:22133
, 100ms wait time, and a single non-concurrent consumer by default.
To use specific options in the base configuration (which supplies configuration to endpoints whose ?properties
are not specified), you can set up a KestrelConfiguration POJO as follows:
<beans xmlns="http://www.springframework.org/schema/beans" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation=" http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd http://camel.apache.org/schema/spring http://camel.apache.org/schema/spring/camel-spring.xsd"> <bean id="kestrelConfiguration" class="org.apache.camel.component.kestrel.KestrelConfiguration"> <property name="addresses" value="kestrel01:22133"/> <property name="waitTimeMs" value="100"/> <property name="concurrentConsumers" value="1"/> </bean> <bean id="kestrel" class="org.apache.camel.component.kestrel.KestrelComponent"> <property name="configuration" ref="kestrelConfiguration"/> </bean> <camelContext xmlns="http://camel.apache.org/schema/spring"> </camelContext> </beans>
Usage Examples
Example 1: Consuming
from("kestrel://kserver02:22133/massive?concurrentConsumers=10&waitTimeMs=500") .bean("myConsumer", "onMessage");
public class MyConsumer { public void onMessage(String message) { ... } }
Example 2: Producing
public class MyProducer { @EndpointInject(uri = "kestrel://kserver01:22133,kserver02:22133/myqueue") ProducerTemplate producerTemplate; public void produceSomething() { producerTemplate.sendBody("Hello, world."); } }
Example 3: Spring XML Configuration
<camelContext xmlns="http://camel.apache.org/schema/spring"> <route> <from uri="kestrel://ks01:22133/sequential?concurrentConsumers=1&waitTimeMs=500"/> <to uri="log:something"/> </route> <route> <from uri="direct:start"/> <to uri="kestrel://ks02:22133/stuff"/> </route> </camelContext>
public class MyBean { public void onMessage(String message) { ... } }
Dependencies
The Kestrel component has the following dependencies:
spymemcached
2.5 (or greater)
spymemcached
You must have the spymemcached
jar on your classpath. Here is a snippet you can use in your pom.xml:
<dependency> <groupId>spy</groupId> <artifactId>memcached</artifactId> <version>2.5</version> </dependency>
Alternatively, you can download the jar directly.