Spring Integration 2.1 introduced support for Redis: "an open source advanced key-value store
".
This support comes in the form of a Redis-based MessageStore
as well as publish-subscribe messaging adapters that are supported by Redis through its PUBLISH
, SUBSCRIBE
, and UNSUBSCRIBE
commands.
To download, install, and run Redis, see the Redis documentation.
To begin interacting with Redis, you first need to connect to it.
Spring Integration uses support provided by another Spring project, Spring Data Redis, which provides typical Spring constructs: ConnectionFactory
and Template
.
Those abstractions simplify integration with several Redis client Java APIs.
Currently Spring Data Redis supports jedis, jredis and Redis Java Client (RJC).
To connect to Redis, you can use one of the implementations of the RedisConnectionFactory
interface.
The following listing shows the interface definition:
public interface RedisConnectionFactory extends PersistenceExceptionTranslator { /** * Provides a suitable connection for interacting with Redis. * * @return connection for interacting with Redis. */ RedisConnection getConnection(); }
The following example shows how to create a JedisConnectionFactory
in Java:
JedisConnectionFactory jcf = new JedisConnectionFactory();
jcf.afterPropertiesSet();
The following example shows how to create a JedisConnectionFactory
in Spring’s XML configuration:
<bean id="redisConnectionFactory" class="o.s.data.redis.connection.jedis.JedisConnectionFactory"> <property name="port" value="7379" /> </bean>
The implementations of RedisConnectionFactory
provide a set of properties, such as port and host, that you can set if needed.
Once you have an instance of RedisConnectionFactory
, you can create an instance of RedisTemplate
and inject it with the RedisConnectionFactory
.
As with other template classes in Spring (such as JdbcTemplate
and JmsTemplate
) RedisTemplate
is a helper class that simplifies Redis data access code.
For more information about RedisTemplate
and its variations (such as StringRedisTemplate
) see the Spring Data Redis documentation.
The following example shows how to create an instance of RedisTemplate
in Java:
RedisTemplate rt = new RedisTemplate<String, Object>();
rt.setConnectionFactory(redisConnectionFactory);
The following example shows how to create an instance of RedisTemplate
in Spring’s XML configuration:
<bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate"> <property name="connectionFactory" ref="redisConnectionFactory"/> </bean>
As mentioned in the introduction, Redis provides support for publish-subscribe messaging through its PUBLISH
, SUBSCRIBE
, and UNSUBSCRIBE
commands.
As with JMS and AMQP, Spring Integration provides message channels and adapters for sending and receiving messages through Redis.
Similarly to JMS, there are cases where both the producer and consumer are intended to be part of the same application, running within the same process. You can accomplished this by using a pair of inbound and outbound channel adapters. However, as with Spring Integration’s JMS support, there is a simpler way to address this use case. You can create a publish-subscribe channel, as the following example shows:
<int-redis:publish-subscribe-channel id="redisChannel" topic-name="si.test.topic"/>
A publish-subscribe-channel
behaves much like a normal <publish-subscribe-channel/>
element from the main Spring Integration namespace.
It can be referenced by both the input-channel
and the output-channel
attributes of any endpoint.
The difference is that this channel is backed by a Redis topic name: a String
value specified by the topic-name
attribute.
However, unlike JMS, this topic does not have to be created in advance or even auto-created by Redis.
In Redis, topics are simple String
values that play the role of an address.
The producer and consumer can communicate by using the same String
value as their topic name.
A simple subscription to this channel means that asynchronous publish-subcribe messaging is possible between the producing and consuming endpoints.
However, unlike the asynchronous message channels created by adding a <queue/>
element within a simple Spring Integration <channel/>
element, the messages are not stored in an in-memory queue.
Instead, those messages are passed through Redis, which lets you rely on its support for persistence and clustering as well as its interoperability with other non-Java platforms.
The Redis inbound channel adapter (RedisInboundChannelAdapter
) adapts incoming Redis messages into Spring messages in the same way as other inbound adapters.
It receives platform-specific messages (Redis in this case) and converts them to Spring messages by using a MessageConverter
strategy.
The following example shows how to configure a Redis inbound channel adapter:
<int-redis:inbound-channel-adapter id="redisAdapter" topics="thing1, thing2" channel="receiveChannel" error-channel="testErrorChannel" message-converter="testConverter" /> <bean id="redisConnectionFactory" class="o.s.data.redis.connection.jedis.JedisConnectionFactory"> <property name="port" value="7379" /> </bean> <bean id="testConverter" class="things.something.SampleMessageConverter" />
The preceding examle shows a simple but complete configuration of a Redis inbound channel adapter.
Note that the preceding configuration relies on the familiar Spring paradigm of auto-discovering certain beans.
In this case, the redisConnectionFactory
is implicitly injected into the adapter.
You can specify it explicitly by using the connection-factory
attribute instead.
Also, note that the preceding configuration injects the adapter with a custom MessageConverter
.
The approach is similar to JMS, where MessageConverter
instances are used to convert between Redis messages and the Spring Integration message payloads.
The default is a SimpleMessageConverter
.
Inbound adapters can subscribe to multiple topic names, hence the comma-separated set of values in the topics
attribute.
Since version 3.0, the inbound adapter, in addition to the existing topics
attribute, now has the topic-patterns
attribute.
This attribute contains a comma-separated set of Redis topic patterns.
For more information regarding Redis publish-subscribe, see Redis Pub/Sub.
Inbound adapters can use a RedisSerializer
to deserialize the body of Redis messages.
The serializer
attribute of the <int-redis:inbound-channel-adapter>
can be set to an empty string, which results in a null
value for the RedisSerializer
property.
In this case, the raw byte[]
bodies of Redis messages are provided as the message payloads.
Since version 5.0, you can provide an Executor
instance to the inbound adapter by using the task-executor
attribute of the <int-redis:inbound-channel-adapter>
.
Also, the received Spring Integration messages now have the RedisHeaders.MESSAGE_SOURCE
header to indicate the source of the published message: topic or pattern.
You can use this downstream for routing logic.
The Redis outbound channel adapter adapts outgoing Spring Integration messages into Redis messages in the same way as other outbound adapters.
It receives Spring Integration messages and converts them to platform-specific messages (Redis in this case) by using a MessageConverter
strategy.
The following example shows how to configure a Redis outbound channel adapter:
<int-redis:outbound-channel-adapter id="outboundAdapter" channel="sendChannel" topic="thing1" message-converter="testConverter"/> <bean id="redisConnectionFactory" class="o.s.data.redis.connection.jedis.JedisConnectionFactory"> <property name="port" value="7379"/> </bean> <bean id="testConverter" class="things.something.SampleMessageConverter" />
The configuration parallels the Redis inbound channel adapter.
The adapter is implicitly injected with a RedisConnectionFactory
, which is defined with redisConnectionFactory
as its bean name.
This example also includes the optional (and custom) MessageConverter
(the testConverter
bean).
Since Spring Integration 3.0, the <int-redis:outbound-channel-adapter>
offers an alternative to the topic
attribute: You can use the topic-expression
attribute to determine the Redis topic for the message at runtime.
These attributes are mutually exclusive.
Spring Integration 3.0 introduced a queue inbound channel adapter to "pop
" messages from a Redis list. By default, it uses "right pop
", but you can configure it to use "left pop
" instead.
The adapter is message-driven.
It uses an internal listener thread and does not use a poller.
The following listing shows all the available attributes for queue-inbound-channel-adapter
:
<int-redis:queue-inbound-channel-adapter id="" channel="" auto-startup="" phase="" connection-factory="" queue="" error-channel="" serializer="" receive-timeout="" recovery-interval="" expect-message="" task-executor="" right-pop=""/>
The component bean name.
If you do not provide the | |
The | |
A | |
A | |
A reference to a | |
The name of the Redis list on which the queue-based pop operation is performed to get Redis messages. | |
The | |
The | |
The timeout in milliseconds for pop operation to wait for a Redis message from the queue. The default is 1 second. | |
The time in milliseconds for which the listener task should sleep after exceptions on the pop operation, before restarting the listener task. | |
Specifies whether this endpoint expects data from the Redis queue to contain entire | |
A reference to a Spring | |
Specifies whether this endpoint should use " |
Spring Integration 3.0 introduced a queue outbound channel adapter to "push
" to a Redis list from Spring Integration messages. By default,
it uses "left push
", but you can configure it to use "right push
" instead.
The following listing shows all the available attributes for a Redis queue-outbound-channel-adapter
:
<int-redis:queue-outbound-channel-adapter id="" channel="" connection-factory="" queue="" queue-expression="" serializer="" extract-payload="" left-push=""/>
The component bean name.
If you do not provide the | |
The | |
A reference to a | |
The name of the Redis list on which the queue-based push operation is performed to send Redis messages.
This attribute is mutually exclusive with | |
A SpEL | |
A | |
Specifies whether this endpoint should send only the payload or the entire | |
Specifies whether this endpoint should use " |
Since Spring Integration 3.0, the Redis module provides an implementation of IntegrationEvent
, which, in turn, is a org.springframework.context.ApplicationEvent
.
The RedisExceptionEvent
encapsulates exceptions from Redis operations (with the endpoint being the "source
" of the event).
For example, the <int-redis:queue-inbound-channel-adapter/>
emits those events after catching exceptions from the BoundListOperations.rightPop
operation.
The exception may be any generic org.springframework.data.redis.RedisSystemException
or a org.springframework.data.redis.RedisConnectionFailureException
.
Handling these events with an <int-event:inbound-channel-adapter/>
can be useful to determine problems with background Redis tasks and to take administrative actions.
As described in the Enterprise Integration Patterns (EIP) book, a message store lets you persist messages.
This can be useful when dealing with components that have a capability to buffer messages (aggregator, resequencer, and others) when reliability is a concern.
In Spring Integration, the MessageStore
strategy also provides the foundation for the claim check pattern, which is described in EIP as well.
Spring Integration’s Redis module provides the RedisMessageStore
.
The following example shows how to use it with a aggregator:
<bean id="redisMessageStore" class="o.s.i.redis.store.RedisMessageStore"> <constructor-arg ref="redisConnectionFactory"/> </bean> <int:aggregator input-channel="inputChannel" output-channel="outputChannel" message-store="redisMessageStore"/>
The preceding example is a bean configuration, and it expects a RedisConnectionFactory
as a constructor argument.
By default, the RedisMessageStore
uses Java serialization to serialize the message.
However, if you want to use a different serialization technique (such as JSON), you can provide your own serializer by setting the valueSerializer
property of the RedisMessageStore
.
Starting with version 4.3.10, the Framework provides Jackson serializer and deserializer implementations for Message
instances and MessageHeaders
instances — MessageJacksonDeserializer
and MessageHeadersJacksonSerializer
, respectively.
They have to be configured with the SimpleModule
options for the ObjectMapper
.
In addition, you should set enableDefaultTyping
on the ObjectMapper
to add type information for each serialized complex object.
That type information is then used during deserialization.
The framework provides a utility method called JacksonJsonUtils.messagingAwareMapper()
, which is already supplied with all the previously mentioned properties and serializers.
To manage JSON serialization in the RedisMessageStore
, you must configure it in a fashion similar to the following example:
RedisMessageStore store = new RedisMessageStore(jedisConnectionFactory); ObjectMapper mapper = JacksonJsonUtils.messagingAwareMapper(); RedisSerializer<Object> serializer = new GenericJackson2JsonRedisSerializer(mapper); store.setValueSerializer(serializer);
Starting with version 4.3.12, RedisMessageStore
supports the prefix
option to allow distinguishing between instances of the store on the same Redis server.
The RedisMessageStore
shown earlier maintains each group as a value under a single key (the group ID).
While you can use this to back a QueueChannel
for persistence, a specialized RedisChannelMessageStore
is provided for that purpose (since version 4.0).
This store uses a LIST
for each channel, LPUSH
when sending messages, and RPOP
when receiving messages.
By default, this store also uses JDK serialization, but you can modify the value serializer, as described earlier.
We recommend using this store backing channels, instead of using the general RedisMessageStore
.
The following example defines a Redis message store and uses it in a channel with a queue:
<bean id="redisMessageStore" class="o.s.i.redis.store.RedisChannelMessageStore"> <constructor-arg ref="redisConnectionFactory"/> </bean> <int:channel id="somePersistentQueueChannel"> <int:queue message-store="redisMessageStore"/> <int:channel>
The keys used to store the data have the form: <storeBeanName>:<channelId>
(in the preceding example, redisMessageStore:somePersistentQueueChannel
).
In addition, a subclass RedisChannelPriorityMessageStore
is also provided.
When you use this with a QueueChannel
, the messages are received in (FIFO) priority order.
It uses the standard IntegrationMessageHeaderAccessor.PRIORITY
header and supports priority values (0 - 9
).
Messages with other priorities (and messages with no priority) are retrieved in FIFO order after any messages with priority.
Important | |
---|---|
These stores implement only |
Spring Integration 3.0 introduced a new Redis-based MetadataStore
(see Section 12.5, “Metadata Store”) implementation.
You can use the RedisMetadataStore
to maintain the state of a MetadataStore
across application restarts.
You can use this new MetadataStore
implementation with adapters such as:
To instruct these adapters to use the new RedisMetadataStore
, declare a Spring bean named metadataStore
.
The Feed inbound channel adapter and the feed inbound channel adapter both automatically pick up and use the declared RedisMetadataStore
.
The following example shows how to declare such a bean:
<bean name="metadataStore" class="o.s.i.redis.store.metadata.RedisMetadataStore"> <constructor-arg name="connectionFactory" ref="redisConnectionFactory"/> </bean>
The RedisMetadataStore
is backed by RedisProperties
.
Interaction with it uses BoundHashOperations
, which, in turn, requires a key
for the entire Properties
store.
In the case of the MetadataStore
, this key
plays the role of a region, which is useful in a distributed environment, when several applications use the same Redis server.
By default, this key
has a value of MetaData
.
Starting with version 4.0, this store implements ConcurrentMetadataStore
, letting it be reliably shared across multiple application instances where only one instance is allowed to store or modify a key’s value.
Important | |
---|---|
You canot use the |
The Redis store inbound channel adapter is a polling consumer that reads data from a Redis collection and sends it as a Message
payload.
The following example shows how to configure a Redis store inbound channel adapter:
<int-redis:store-inbound-channel-adapter id="listAdapter" connection-factory="redisConnectionFactory" key="myCollection" channel="redisChannel" collection-type="LIST" > <int:poller fixed-rate="2000" max-messages-per-poll="10"/> </int-redis:store-inbound-channel-adapter>
The preceding example shows how to configure a Redis store inbound channel adapter by using the store-inbound-channel-adapter
element, providing values for various attributes, such as:
key
or key-expression
: The name of the key for the collection being used.
collection-type
: An enumeration of the collection types supported by this adapter.
The supported Collections are LIST
, SET
, ZSET
, PROPERTIES
, and MAP
.
connection-factory
: Reference to an instance of o.s.data.redis.connection.RedisConnectionFactory
.
redis-template
: Reference to an instance of o.s.data.redis.core.RedisTemplate
.
Note | |
---|---|
You cannot set both |
Important | |
---|---|
By default, the adapter uses a <bean id="redisTemplate" class="org.springframework.data.redis.core.RedisTemplate"> <property name="connectionFactory" ref="redisConnectionFactory"/> <property name="keySerializer"> <bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/> </property> <property name="hashKeySerializer"> <bean class="org.springframework.data.redis.serializer.StringRedisSerializer"/> </property> </bean> Th |
Because it has a literal value for the key
, the preceding example is relatively simple and static.
Sometimes, you may need to change the value of the key at runtime based on some condition.
To do so, use key-expression
instead, where the provided expression can be any valid SpEL expression.
Also, you may wish to perform some post-processing on the successfully processed data that was read from the Redis collection.
For example, you may want to move or remove the value after its been processed.
You can do so by using the transaction synchronization feature that was added with Spring Integration 2.2.
The following example uses key-expression
and transaction synchronization:
<int-redis:store-inbound-channel-adapter id="zsetAdapterWithSingleScoreAndSynchronization" connection-factory="redisConnectionFactory" key-expression="'presidents'" channel="otherRedisChannel" auto-startup="false" collection-type="ZSET"> <int:poller fixed-rate="1000" max-messages-per-poll="2"> <int:transactional synchronization-factory="syncFactory"/> </int:poller> </int-redis:store-inbound-channel-adapter> <int:transaction-synchronization-factory id="syncFactory"> <int:after-commit expression="payload.removeByScore(18, 18)"/> </int:transaction-synchronization-factory> <bean id="transactionManager" class="o.s.i.transaction.PseudoTransactionManager"/>
You can declare your poller to be transactional by using a transactional
element.
This element can reference a real transaction manager (for example, if some other part of your flow invokes JDBC).
If you do not have a "real
" transaction, you can use an o.s.i.transaction.PseudoTransactionManager
, which is an implementation of Spring’s PlatformTransactionManager
and enables the use of the transaction synchronization features of the Redis adapter when there is no actual transaction.
Important | |
---|---|
This does not make the Redis activities themselves transactional. It lets the synchronization of actions be taken before or after success (commit) or after failure (rollback). |
Once your poller is transactional, you can set an instance of the o.s.i.transaction.TransactionSynchronizationFactory
on the transactional
element.
TransactionSynchronizationFactory
creates an instance of the TransactionSynchronization
.
For your convenience, we have exposed a default SpEL-based TransactionSynchronizationFactory
, which lets you configure SpEL expressions, with their execution being coordinated (synchronized) with a transaction.
Expressions for before-commit, after-commit, and after-rollback are supported, together with channels (one for each kind of event) where the evaluation result (if any) is sent.
For each child element, you can specify expression
and channel
attributes.
If only the channel
attribute is present, the received message is sent there as part of the particular synchronization scenario.
If only the expression
attribute is present and the result of an expression is a non-null value, a message with the result as the payload is generated and sent to a default channel (NullChannel
) and appears in the logs (at the DEBUG
level).
If you want the evaluation result to go to a specific channel, add a channel
attribute.
If the result of an expression is null or void, no message is generated.
For more information about transaction synchronization, see Section C.3, “Transaction Synchronization”.
The RedisStore outbound channel adapter lets you write a message payload to a Redis collection, as the following example shows:
<int-redis:store-outbound-channel-adapter id="redisListAdapter" collection-type="LIST" channel="requestChannel" key="myCollection" />
The preceding configuration a Redis store outbound channel adapter by using the store-inbound-channel-adapter
element.
It provides values for various attributes, such as:
key
or key-expression
: The name of the key for the collection being used.
extract-payload-elements
: If set to true
(the default) and the payload is an instance of a "multi-value
" object (that is, a Collection
or a Map
), it is stored by using "addAll
" and "putAll
" semantics.
Otherwise, if set to false
, the payload is stored as a single entry regardless of its type.
If the payload is not an instance of a "multi-value
" object, the value of this attribute is ignored and the payload is always stored as a single entry.
collection-type
: An enumeration of the Collection
types supported by this adapter.
The supported Collections are LIST
, SET
, ZSET
, PROPERTIES
, and MAP
.
map-key-expression
: SpEL expression that returns the name of the key for the entry being stored.
It applies only if the collection-type
is MAP
or PROPERTIES
and extract-payload-elements is false.
connection-factory
: Reference to an instance of o.s.data.redis.connection.RedisConnectionFactory
.
redis-template
: Reference to an instance of o.s.data.redis.core.RedisTemplate
.
Note | |
---|---|
You cannot set both |
Important | |
---|---|
By default, the adapter uses a |
Because it has literal values for the key
and other attributes, the preceding example is relatively simple and static.
Sometimes, you may need to change the values dynamically at runtime based on some condition.
To do so, use their -expression
equivalents (key-expression
, map-key-expression
, and so on), where the provided expression can be any valid SpEL expression.
Spring Integration 4.0 introduced the Redis command gateway to let you perform any standard Redis command by using the generic RedisConnection#execute
method.
The following listing shows the available attributes for the Redis outbound gateway:
<int-redis:outbound-gateway request-channel="" reply-channel="" requires-reply="" reply-timeout="" connection-factory="" redis-template="" arguments-serializer="" command-expression="" argument-expressions="" use-command-variable="" arguments-strategy="" />
The | |
The | |
Specifies whether this outbound gateway must return a non-null value.
It defaults to | |
The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels. | |
A reference to a | |
A reference to a | |
A reference to an instance of | |
The SpEL expression that returns the command key.
It defaults to the | |
Comma-separated SpEL expressions that are evaluated as command arguments.
Mutually exclusive with the | |
A | |
Reference to an instance of |
You can use the <int-redis:outbound-gateway>
as a common component to perform any desired Redis operation.
For example, the following examlpe shows how to get incremented values from Redis atomic number:
<int-redis:outbound-gateway request-channel="requestChannel" reply-channel="replyChannel" command-expression="'INCR'"/>
The Message
payload should have a name of redisCounter
, which may be provided by org.springframework.data.redis.support.atomic.RedisAtomicInteger
bean definition.
The RedisConnection#execute
method has a generic Object
as its return type.
Real result depends on command type.
For example, MGET
returns a List<byte[]>
.
For more information about commands, their arguments and result type, see Redis Specification.
Spring Integration introduced the Redis queue outbound gateway to perform request and reply scenarios.
It pushes a conversation UUID
to the provided queue
, pushes the value with that UUID
as its key to a Redis list, and waits for the reply from a Redis list with a key of UUID' plus '.reply
.
A different UUID is used for each interaction.
The following listing shows the available attributes for a Redis outbound gateway:
<int-redis:queue-outbound-gateway request-channel="" reply-channel="" requires-reply="" reply-timeout="" connection-factory="" queue="" order="" serializer="" extract-payload=""/>
The | |
The | |
Specifies whether this outbound gateway must return a non-null value.
This value is | |
The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels. | |
A reference to a | |
The name of the Redis list to which the outbound gateway sends a conversation | |
The order of this outbound gateway when multiple gateways are registered. | |
The | |
Specifies whether this endpoint expects data from the Redis queue to contain entire |
Spring Integration 4.1 introduced the Redis queue inbound gateway to perform request and reply scenarios.
It pops a conversation UUID
from the provided queue
, pops the value with that UUID
as its key from the Redis list, and pushes the reply to the Redis list with a key of UUID
plus .reply
.
The following listing shows the available attributes for a Redis queue inbound gateway:
<int-redis:queue-inbound-gateway request-channel="" reply-channel="" executor="" reply-timeout="" connection-factory="" queue="" order="" serializer="" receive-timeout="" expect-message="" recovery-interval=""/>
The | |
The | |
A reference to a Spring | |
The timeout (in milliseconds) to wait until the reply message is sent. It is typically applied for queue-based limited reply-channels. | |
A reference to a | |
The name of the Redis list for the conversation | |
The order of this inbound gateway when multiple gateways are registered. | |
The | |
The timeout (in milliseconds) to wait until the receive message is fetched. It is typically applied for queue-based limited request-channels. | |
Specifies whether this endpoint expects data from the Redis queue to contain entire | |
The time (in milliseconds) the listener task should sleep after exceptions on the " |
Spring Integration 4.0 introduced the RedisLockRegistry
.
Certain components (for example, aggregator and resequencer) use a lock obtained from a LockRegistry
instance to ensure that only one thread manipulates a group at a time.
The DefaultLockRegistry
performs this function within a single component.
You can now configure an external lock registry on these components.
When you use it with a shared MessageGroupStore
, you can use the RedisLockRegistry
to provide this functionality across multiple application instances, such that only one instance can manipulate the group at a time.
When a lock is released by a local thread, another local thread can generally acquire the lock immediately. If a lock is released by a thread using a different registry instance, it can take up to 100ms to acquire the lock.
To avoid "hung
" locks (when a server fails), the locks in this registry are expired after a default 60 seconds, but you can configure this value on the registry.
Locks are normally held for a much smaller time.
Important | |
---|---|
Because the keys can expire, an attempt to unlock an expired lock results in an exception being thrown. However, the resources protected by such a lock may have been compromised, so such exceptions should be considered to be severe. You should set the expiry at a large enough value to prevent this condition, but set it low enough that the lock can be recovered after a server failure in a reasonable amount of time. |
Starting with version 5.0, the RedisLockRegistry
implements ExpirableLockRegistry
, which removes locks last acquired more than age
ago and that are not currently locked.