Showing posts with label redis. Show all posts
Showing posts with label redis. Show all posts

Monday, January 9, 2017

Overcast and Docker: test your apps as if you are releasing them to production

I am sure everyone would agree that rare application architecture these days could survive without relying on some kind of data stores (either relational or NoSQL ones), or/and messaging middleware, or/and external caches, ... just to name a few. Testing such applications becomes a real challenge.

Luckily, if you are a JVM developer, things are not as bad. Most of the time you have an option to fallback to an embedded version of the data store or message broker in your integration or component test scenarios. But what if the solution you are using is not JVM-based? Great examples of those are RabbitMQ, Redis, Memcached, MySQL, Postgresql which are extremely popular choices these days, and for a very good reasons. Even better, what if your integration testing strategy is set to exercise the component (read, microservice) in the environment as close to production as possible? Should we give up here? Or should we write a bunch of flaky shell scripts to orchestrate the test runs and scary most of the developers to death? Let us see what we can do here ...

Many of you are already screaming at this point: just use Docker, or CoreOS! And this is exactly what we are going to talk about in this post, more precisely, how to use Docker to back integration / component testing. I think Docker does not need an introduction anymore. Even those of us who spent a last couple of years in a cave on a deserted island in the middle of the ocean have heard about it.

Our sample application is going to be built on top of Spring projects portfolio, heavily relying on Spring Boot magic to wire all the pieces together (who doesn't, right? it works pretty well indeed). It will be implementing a very simple workflow: publish a message to RabbitMQ exchange app.exchange (using app.queue routing key), consume the message from the RabbitMQ queue app.queue and store it in Redis list data structure under the key messages. The three self-explanatory code snippets below demonstrate how each functional piece is being done:

@Component
public class AppQueueMessageSender {
    @Autowired private RabbitTemplate rabbitTemplate;
 
    public void send(final String message) {
        rabbitTemplate.convertAndSend("app.exchange", "app.queue", message);
    }
}
@Component
public class AppQueueMessageListener {
    @Autowired private AppMessageRepository repository;
 
    @RabbitListener(
        queues = "app.queue", 
        containerFactory = "rabbitListenerContainerFactory", 
        admin = "amqpAdmin"
    )
    public void onMessage(final String message) {
        repository.persist(message);
    }
}
@Repository
public class AppMessageRepository {
    @Autowired private StringRedisTemplate redisTemplate;
 
    public void persist(final String message) {
        redisTemplate.opsForList().rightPush("messages", message);
    }
 
    public Collection<String> readAll() {
        return redisTemplate.opsForList().range("messages", 0, -1);
    }
 
    public long size() {
        return redisTemplate.opsForList().size("messages");
    }
}

As you can see, the implementation is deliberately doing the bare minimum, we are more interested in the fact that quite a few interactions with RabbitMQ and Redis are happening here. The configuration class includes only the necessary beans, everything else has been figured out by Spring Boot automatic discovery from the classpath dependencies.

@Configuration
@EnableAutoConfiguration
@EnableRabbit
@ComponentScan(basePackageClasses = AppConfiguration.class)
public class AppConfiguration {
    @Bean
    Queue queue() {
        return new Queue("app.queue", false);
    }

    @Bean
    TopicExchange exchange() {
        return new TopicExchange("app.exchange");
    }

    @Bean
    Binding binding(Queue queue, TopicExchange exchange) {
        return BindingBuilder.bind(queue).to(exchange).with(queue.getName());
    }
 
    @Bean
    StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
        return new StringRedisTemplate(connectionFactory);
    }
}

At the very end comes application.yml. Essentially it contains the default connection parameters for RabbitMQ and Redis, plus a bit of logging level tuning.

logging:
  level:
    root: INFO
    
spring:
  rabbitmq:
    host: localhost
    username: guest
    password: guest
    port: 5672
    
  redis:
    host: localhost
    port: 6379

With that, our application is ready be run. For convenience, the project repository contains docker-compose.yml with official RabbitMQ and Redis images from the Docker hub.

Being TDD believers and practitioners, we make sure no application leaves the gate without thorough set of test suites and test cases. Keeping unit tests and integration tests out of scope of the subject of our discussion, let us jump right into component testing with this simple scenario.

@RunWith(SpringJUnit4ClassRunner.class)
@SpringBootTest(
    classes = { AppConfiguration.class }, 
    webEnvironment = WebEnvironment.NONE
)
public class AppComponentTest {
    @Autowired private AppQueueMessageSender sender;
    @Autowired private AppMessageRepository repository;
 
    @Test
    public void testThatMessageHasBeenPersisted() {
        sender.send("Test Message!");
        await().atMost(1, SECONDS).until(() -> repository.size() > 0);
        assertThat(repository.readAll()).containsExactly("Test Message!");
    }
}

It is really basic test case, exercising the main flow, but it makes an important point: no mocks / stubs / ... allowed, we expect the real things. The line is somewhat blurry but this is what makes component tests different from let say integration or e2e tests: we test a single full-fledged application (component) with the real dependencies (when it makes sense).

This is a right time for an excellent Overcast project to appear on the stage and help us out. Overcast brings the power of Docker to enrich test harness of the JVM applications. Among many other things, it allows to define and manage lifecycle of Docker containers from within Java code (or more precisely, any programming language based on JVM).

Unfortunately, the last released version 2.5.1 of Overcast is pretty old and does not include a lot of features and enhancements. However, building it from source is no-brainer (hopefully the new release is going to be available soon).

git clone https://github.com/xebialabs/overcast
cd overcast
./gradlew install

Essentially, the only prerequisite is to provide the configuration file overcast.conf with the list of named containers to run. In our case, we need RabbitMQ and Redis.

rabbitmq {
    dockerImage="rabbitmq:3.6.6"
    exposeAllPorts=true
    remove=true
    removeVolume=true
}

redis {
    dockerImage="redis:3.2.6"
    exposeAllPorts=true
    remove=true
    removeVolume=true
}

Great! The syntax is not as powerful as Docker Compose supports, but simple, straightforward and quite sufficient to be fair. Once configuration file is placed into src/test/resources folder, we could move on and use Overcast Java API to manage these containers programmatically. It is natural to introduce the dedicated configuration class in this case as we are using Spring framework.

@Configuration
public class OvercastConfiguration {
    @Autowired private ConfigurableEnvironment env;

    @Bean(initMethod = "setup", destroyMethod = "teardown")
    @Qualifier("rabbitmq")
    public CloudHost rabbitmq() {
        return CloudHostFactory.getCloudHost("rabbitmq");
    }

    @Bean(initMethod = "setup", destroyMethod = "teardown")
    @Qualifier("redis")
    public CloudHost redis() {
        return CloudHostFactory.getCloudHost("redis");
    }

    @PostConstruct
    public void init() throws TimeoutException {
        final CloudHost redis = redis();
        final CloudHost rabbitmq = rabbitmq();

        final Map<String, Object> properties = new HashMap<>();
        properties.put("spring.rabbitmq.host", rabbitmq.getHostName());
        properties.put("spring.rabbitmq.port", rabbitmq.getPort(5672));

        properties.put("spring.redis.host", redis.getHostName());
        properties.put("spring.redis.port", redis.getPort(6379));

        final PropertySource<?> source = new MapPropertySource("overcast", properties);
        env.getPropertySources().addFirst(source);
    }
}

And that is literally all we need! Just a couple of important notes here. Docker is going to expose random ports for each container, so we could run many test cases in parallel on the same box without any port conflicts. On most operating systems it is safe to use localhost to access the running containers but for the ones without native Docker support the workarounds with Docker Machine or boot2docker exist. That is why we override connection settings for both host and port for RabbitMQ and Redis respectively, asking for the bindings at runtime:

properties.put("spring.rabbitmq.host", rabbitmq.getHostName());
properties.put("spring.rabbitmq.port", rabbitmq.getPort(5672));

properties.put("spring.redis.host", redis.getHostName());
properties.put("spring.redis.port", redis.getPort(6379));

Lastly, more advanced Docker users may wonder how Overcast is able to figure out where Docker daemon is running? Which port it is bound to? Does it use TLS or not? Under the hood Overcast uses terrific Spotify Docker Client which is able to retrieve all the relevant details from the environment variables, which works in majority of use cases (though you can always provide your own settings).

To finish up, let us include this configuration into the test case:

@SpringBootTest(
    classes = { OvercastConfiguration.class, AppConfiguration.class }, 
    webEnvironment = WebEnvironment.NONE
)

Easy, isn't it? If we go ahead and run mvn test for our project, all test cases should pass (please notice that first run may take some time as Docker would have to pull the container images from the remote repository).

> mvn test
...
Tests run: 1, Failures: 0, Errors: 0, Skipped: 0
...

No doubts, Docker raises testing techniques to a new level. With the help of such awesome libraries as Overcast, seasoned JVM developers have even more options to come up with realistic test scenarios and run them against the components in "mostly" production environment (on the wave of hype, it fits perfectly into microservices testing strategies). There are many areas where Overcast could and will improve but it brings a lot of value even now, definitely worth checking out.

Probably, the most annoying issue you may encounter when working with Docker containers is awaiting for the moment when the container is fully started and ready to accept the requests (which heavily depends on what kind of underlying service this container is running). Although the work on that has been started, Overcast does not help with this particular problem yet though simple, old-style sleeps may be good enough (versus a bit more complex port polling for example).

But but but ... always remember about testing pyramid and strive for a right balance. Create as many test cases as you need to cover most critical and important flows, but no more. Unit and integration tests should be your main weapon.

The complete project is available on Github.

Sunday, February 23, 2014

Knowing how all your components work together: distributed tracing with Zipkin

In today's post we will try to cover very interesting and important topic: distributed system tracing. What it practically means is that we will try to trace the request from the point it was issued by the client to the point the response to this request was received. At first, it looks quite straightforward but in reality it may involve many calls to several other systems, databases, NoSQL stores, caches, you name it ...

In 2010 Google published a paper about Dapper, a large-scale distributed systems tracing infrastructure (very interesting reading by the way). Later on, Twitter built its own implementation based on Dapper paper, called Zipkin and that's the one we are going to look at.

We will build a simple JAX-RS 2.0 server using great Apache CXF library. For the client side, we will use JAX-RS 2.0 client API and by utilizing Zipkin we will trace all the interactions between the client and the server (as well as everything happening on server side). To make an example a bit more illustrative, we will pretend that server uses some kind of database to retrieve the data. Our code will be a mix of pure Java and a bit of Scala (the choice of Scala will be cleared up soon).

One additional dependency in order for Zipkin to work is Apache Zookeeper. It is required for coordination and should be started in advance. Luckily, it is very easy to do:

  • download the release from http://zookeeper.apache.org/releases.html (the current stable version at the moment of writing is 3.4.5)
  • unpack it into zookeeper-3.4.5
  • copy zookeeper-3.4.5/conf/zoo_sample.cfg to zookeeper-3.4.5/conf/zoo.cfg
  • and just start Apache Zookeeper server
    Windows: zookeeper-3.4.5/bin/zkServer.cmd
    Linux: zookeeper-3.4.5/bin/zkServer.sh start

Now back to Zipkin. Zipkin is written in Scala. It is still in active development and the best way to start off with it is just by cloning its GitHub repository and build it from sources:

git clone https://github.com/twitter/zipkin.git

From architectural prospective, Zipkin consists of three main components:

  • collector: collects traces across the system
  • query: queries collected traces
  • web: provides web-based UI to show the traces

To run them, Zipkin guys provide useful scripts in the bin folder with the only requirement that JDK 1.7 should be installed:

  • bin/collector
  • bin/query
  • bin/web
Let's execute these scripts and ensure that every component has been started successfully, with no stack traces on the console (for curious readers, I was not able to make Zipkin work on Windows so I assume we are running it on Linux box). By default, Zipkin web UI is available on port 8080. The storage for traces is embedded SQLite engine. Though it works, the better storages (like awesome Redis) are available.

The preparation is over, let's do some code. We will start with JAX-RS 2.0 client part as it's very straightforward (ClientStarter.java):

package com.example.client;

import javax.ws.rs.client.Client;
import javax.ws.rs.client.ClientBuilder;
import javax.ws.rs.core.MediaType;
import javax.ws.rs.core.Response;

import com.example.zipkin.Zipkin;
import com.example.zipkin.client.ZipkinRequestFilter;
import com.example.zipkin.client.ZipkinResponseFilter;

public class ClientStarter {
  public static void main( final String[] args ) throws Exception { 
    final Client client = ClientBuilder
      .newClient()
      .register( new ZipkinRequestFilter( "People", Zipkin.tracer() ), 1 )
      .register( new ZipkinResponseFilter( "People", Zipkin.tracer() ), 1 );        
                        
    final Response response = client
      .target( "http://localhost:8080/rest/api/people" )
      .request( MediaType.APPLICATION_JSON )
      .get();

    if( response.getStatus() == 200 ) {
      System.out.println( response.readEntity( String.class ) );
    }
        
    response.close();
    client.close();
        
    // Small delay to allow tracer to send the trace over the wire
    Thread.sleep( 1000 );
  }
}

Except a couple of imports and classes with Zipkin in it, everything should look simple. So what those ZipkinRequestFilter and ZipkinResponseFilter are for? Zipkin is awesome but it's not a magical tool. In order to trace any request in distributed system, there should be some context passed along with it. In REST/HTTP world, it's usually request/response headers. Let's take a look on ZipkinRequestFilter first (ZipkinRequestFilter.scala):

package com.example.zipkin.client

import javax.ws.rs.client.ClientRequestFilter
import javax.ws.rs.ext.Provider
import javax.ws.rs.client.ClientRequestContext
import com.twitter.finagle.http.HttpTracing
import com.twitter.finagle.tracing.Trace
import com.twitter.finagle.tracing.Annotation
import com.twitter.finagle.tracing.TraceId
import com.twitter.finagle.tracing.Tracer

@Provider
class ZipkinRequestFilter( val name: String, val tracer: Tracer ) extends ClientRequestFilter {
  def filter( requestContext: ClientRequestContext ): Unit = {      
    Trace.pushTracerAndSetNextId( tracer, true )
 
    requestContext.getHeaders().add( HttpTracing.Header.TraceId, Trace.id.traceId.toString )
    requestContext.getHeaders().add( HttpTracing.Header.SpanId, Trace.id.spanId.toString )
         
    Trace.id._parentId foreach { id => 
      requestContext.getHeaders().add( HttpTracing.Header.ParentSpanId, id.toString ) 
    }    

    Trace.id.sampled foreach { sampled => 
      requestContext.getHeaders().add( HttpTracing.Header.Sampled, sampled.toString ) 
    }

    requestContext.getHeaders().add( HttpTracing.Header.Flags, Trace.id.flags.toLong.toString )
             
    if( Trace.isActivelyTracing ) {
      Trace.recordRpcname( name,  requestContext.getMethod() )
      Trace.recordBinary( "http.uri", requestContext.getUri().toString()  )
      Trace.record( Annotation.ClientSend() )    
    }
  }
}

A bit of Zipkin internals will make this code superclear. The central part of Zipkin API is Trace class. Every time we would like to initiate tracing, we should have a Trace Id and the tracer to actually trace it. This single line generates new Trace Id and register the tracer (internally this data is held in thread local state).

Trace.pushTracerAndSetNextId( tracer, true )

Traces are hierarchical by nature, so do Trace Ids: every Trace Id could be a root or part of another trace. In our example, we know for sure that we are the first and as such the root of the trace. Later on the Trace Id is wrapped into HTTP headers and will be passed along the request (we will see on server side how it is being used). The last three lines associate the useful information with the trace: name of our API (People), HTTP method, URI and most importantly, that it's the client sending the request to the server.

Trace.recordRpcname( name,  requestContext.getMethod() )
Trace.recordBinary( "http.uri", requestContext.getUri().toString()  )
Trace.record( Annotation.ClientSend() ) 

The ZipkinResponseFilter does the reverse to ZipkinRequestFilter and extract the Trace Id from the request headers (ZipkinResponseFilter.scala):

package com.example.zipkin.client

import javax.ws.rs.client.ClientResponseFilter
import javax.ws.rs.client.ClientRequestContext
import javax.ws.rs.client.ClientResponseContext
import javax.ws.rs.ext.Provider
import com.twitter.finagle.tracing.Trace
import com.twitter.finagle.tracing.Annotation
import com.twitter.finagle.tracing.SpanId
import com.twitter.finagle.http.HttpTracing
import com.twitter.finagle.tracing.TraceId
import com.twitter.finagle.tracing.Flags
import com.twitter.finagle.tracing.Tracer

@Provider
class ZipkinResponseFilter( val name: String, val tracer: Tracer ) extends ClientResponseFilter {  
  def filter( requestContext: ClientRequestContext, responseContext: ClientResponseContext ): Unit = {
    val spanId = SpanId.fromString( requestContext.getHeaders().getFirst( HttpTracing.Header.SpanId ).toString() )

    spanId foreach { sid =>
      val traceId = SpanId.fromString( requestContext.getHeaders().getFirst( HttpTracing.Header.TraceId ).toString() )
        
      val parentSpanId = requestContext.getHeaders().getFirst( HttpTracing.Header.ParentSpanId ) match {
        case s: String => SpanId.fromString( s.toString() )
        case _ => None
      }

      val sampled = requestContext.getHeaders().getFirst( HttpTracing.Header.Sampled ) match { 
        case s: String =>  s.toString.toBoolean
        case _ => true
      }
        
      val flags = Flags( requestContext.getHeaders().getFirst( HttpTracing.Header.Flags ).toString.toLong )        
      Trace.setId( TraceId( traceId, parentSpanId, sid, Option( sampled ), flags ) )
    }
      
    if( Trace.isActivelyTracing ) {
      Trace.record( Annotation.ClientRecv() )
    }
  }
}

Strictly speaking, in our example it's not necessary to extract the Trace Id from the request because both filters should be executed by the single thread. But the last line is very important: it marks the end of our trace by saying that client has received the response.

Trace.record( Annotation.ClientRecv() )

What's left is actually the tracer itself (Zipkin.scala):

package com.example.zipkin

import com.twitter.finagle.stats.DefaultStatsReceiver
import com.twitter.finagle.zipkin.thrift.ZipkinTracer
import com.twitter.finagle.tracing.Trace
import javax.ws.rs.ext.Provider

object Zipkin {
  lazy val tracer = ZipkinTracer.mk( host = "localhost", port = 9410, DefaultStatsReceiver, 1 )
}

If at this point you are confused what all those traces and spans mean please look through this documentation page, you will get the basic understanding of those concepts.

At this point, there is nothing left on the client side and we are good to move to the server side. Our JAX-RS 2.0 server will expose the single endpoint (PeopleRestService.java):

package com.example.server.rs;

import java.util.Arrays;
import java.util.Collection;
import java.util.concurrent.Callable;

import javax.ws.rs.GET;
import javax.ws.rs.Path;
import javax.ws.rs.Produces;

import com.example.model.Person;
import com.example.zipkin.Zipkin;

@Path( "/people" ) 
public class PeopleRestService {
  @Produces( { "application/json" } )
  @GET
  public Collection< Person > getPeople() {
    return Zipkin.invoke( "DB", "FIND ALL", new Callable< Collection< Person > >() {
      @Override
      public Collection call() throws Exception {
        return Arrays.asList( new Person( "Tom", "Bombdil" ) );
      }   
    } );   
  }
}

As we mentioned before, we will simulate the access to database and generate a child trace by using Zipkin.invoke wrapper (which looks very simple, Zipkin.scala):

package com.example.zipkin

import java.util.concurrent.Callable
import com.twitter.finagle.stats.DefaultStatsReceiver
import com.twitter.finagle.tracing.Trace
import com.twitter.finagle.zipkin.thrift.ZipkinTracer
import com.twitter.finagle.tracing.Annotation

object Zipkin {
  lazy val tracer = ZipkinTracer.mk( host = "localhost", port = 9410, DefaultStatsReceiver, 1 )
    
  def invoke[ R ]( service: String, method: String, callable: Callable[ R ] ): R = Trace.unwind {
    Trace.pushTracerAndSetNextId( tracer, false )      
      
    Trace.recordRpcname( service, method );
    Trace.record( new Annotation.ClientSend() );
          
    try {
      callable.call()
    } finally {
      Trace.record( new Annotation.ClientRecv() );
    }
  }   
}

As we can see, in this case the server itself becomes a client for some other service (database).

The last and most important part of the server is to intercept all HTTP requests, extract the Trace Id from them so it will be possible to associate more data with the trace (annotate the trace). In Apache CXF it's very easy to do by providing own invoker (ZipkinTracingInvoker.scala):

package com.example.zipkin.server

import org.apache.cxf.jaxrs.JAXRSInvoker
import com.twitter.finagle.tracing.TraceId
import org.apache.cxf.message.Exchange
import com.twitter.finagle.tracing.Trace
import com.twitter.finagle.tracing.Annotation
import org.apache.cxf.jaxrs.model.OperationResourceInfo
import org.apache.cxf.jaxrs.ext.MessageContextImpl
import com.twitter.finagle.tracing.SpanId
import com.twitter.finagle.http.HttpTracing
import com.twitter.finagle.tracing.Flags
import scala.collection.JavaConversions._
import com.twitter.finagle.tracing.Tracer
import javax.inject.Inject

class ZipkinTracingInvoker extends JAXRSInvoker {
  @Inject val tracer: Tracer = null
  
  def trace[ R ]( exchange: Exchange )( block: => R ): R = {
    val context = new MessageContextImpl( exchange.getInMessage() )
    Trace.pushTracer( tracer )
        
    val id = Option( exchange.get( classOf[ OperationResourceInfo ] ) ) map { ori =>
      context.getHttpHeaders().getRequestHeader( HttpTracing.Header.SpanId ).toList match {
        case x :: xs => SpanId.fromString( x ) map { sid => 
          val traceId = context.getHttpHeaders().getRequestHeader( HttpTracing.Header.TraceId ).toList match {
            case x :: xs => SpanId.fromString( x )
            case _ => None
          }
          
          val parentSpanId = context.getHttpHeaders().getRequestHeader( HttpTracing.Header.ParentSpanId ).toList match {
            case x :: xs => SpanId.fromString( x )
            case _ => None
          }
  
          val sampled = context.getHttpHeaders().getRequestHeader( HttpTracing.Header.Sampled ).toList match { 
            case x :: xs =>  x.toBoolean
            case _ => true
          }
                    
          val flags = context.getHttpHeaders().getRequestHeader( HttpTracing.Header.Flags ).toList match {
            case x :: xs =>  Flags( x.toLong )
            case _ => Flags()
          }
         
          val id = TraceId( traceId, parentSpanId, sid, Option( sampled ), flags )                     
          Trace.setId( id )
        
          if( Trace.isActivelyTracing ) {
            Trace.recordRpcname( context.getHttpServletRequest().getProtocol(), ori.getHttpMethod() )
            Trace.record( Annotation.ServerRecv() )
          }
        
          id
        }           
          
        case _ => None
      }
    }
    
    val result = block
    
    if( Trace.isActivelyTracing ) {
      id map { id => Trace.record( new Annotation.ServerSend() ) }
    }
    
    result
  }
  
  @Override
  override def invoke( exchange: Exchange, parametersList: AnyRef ): AnyRef = {
    trace( exchange )( super.invoke( exchange, parametersList ) )     
  }
}

Basically, the only thing this code does is extracting Trace Id from request and associating it with the current thread. Also please notice that we associate additional data with the trace marking the server participation.

  Trace.recordRpcname( context.getHttpServletRequest().getProtocol(), ori.getHttpMethod() )
  Trace.record( Annotation.ServerRecv() )

To see the tracing in live, let's start our server (please notice that sbt should be installed), assuming all Zipkin components and Apache Zookeeper are already up and running:

sbt 'project server' 'run-main com.example.server.ServerStarter'
then the client:
sbt 'project client' 'run-main com.example.client.ClientStarter'
and finally open Zipkin web UI at http://localhost:8080. We should see something like that (depending how many times you have run the client):

Alternatively, we can build and run fat JARs using sbt-assembly plugin:

sbt assembly
java -jar server/target/zipkin-jaxrs-2.0-server-assembly-0.0.1-SNAPSHOT.jar 
java -jar client/target/zipkin-jaxrs-2.0-client-assembly-0.0.1-SNAPSHOT.jar 

If we click on any particular trace, the more detailed information will be shown, much resembling client <-> server <-> database chain.

Even more details are shown when we click on particular element in the tree.

Lastly, the bonus part is components / services dependency graph.

As we can see, all the data associated with the trace is here and follows hierarchical structure. The root and child traces are detected and shown, as well as timelines for client send/receive and server receive/send chains. Our example is quite naive and simple, but even like that it demonstrates how powerful and useful distributed system tracing is. Thanks to Zipkin guys.

The complete source code is available on GitHub.

Sunday, September 30, 2012

Redis pub/sub using Spring

Continuing to discover the powerful set of Redis features, the one worth mentioning about is out of the box support of pub/sub messaging.

Pub/Sub messaging is essential part of many software architectures. Some software systems demand from messaging solution to provide high-performance, scalability, queues persistence and durability, fail-over support, transactions, and many more nice-to-have features, which in Java world mostly always leads to using one of JMS implementation providers. In my previous projects I have actively used Apache ActiveMQ (now moving towards Apache ActiveMQ Apollo). Though it's a great implementation, sometimes I just needed simple queuing support and Apache ActiveMQ just looked overcomplicated for that.

Alternatives? Please welcome Redis pub/sub! If you are already using Redis as key/value store, few additional lines of configuration will bring pub/sub messaging to your application in no time.

Spring Data Redis project abstracts very well Redis pub/sub API and provides the model so familiar to everyone who uses Spring capabilities to integrate with JMS.

As always, let's start with the POM configuration file. It's pretty small and simple, includes necessary Spring dependencies, Spring Data Redis and Jedis, great Java client for Redis.



    4.0.0
    com.example.spring
    redis
    0.0.1-SNAPSHOT
    jar

    
        UTF-8
        3.1.1.RELEASE
    

    
        
            org.springframework.data
            spring-data-redis
            1.0.1.RELEASE
        

        
            cglib
            cglib-nodep
            2.2
        

        
            log4j
            log4j
            1.2.16
        

        
            redis.clients
            jedis
            2.0.0
            jar
        

        
            org.springframework
            spring-core
            ${spring.version}
        

        
            org.springframework
            spring-context
            ${spring.version}
           
       

       
           
               
                   org.apache.maven.plugins
                   maven-compiler-plugin
                   2.3.2
                   
                       1.6
                       1.6
                   
               
           
    


Moving on to configuring Spring context, let's understand what we need to have in order for a publisher to publish some messages and for a consumer to consume them. Knowing the respective Spring abstractions for JMS will help a lot with that.

  • we need connection factory -> JedisConnectionFactory
  • we need a template for publisher to publish messages -> RedisTemplate
  • we need a message listener for consumer to consume messages -> RedisMessageListenerContainer
Using Spring Java configuration, let's describe our context:
package com.example.redis.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.scheduling.annotation.EnableScheduling;

import com.example.redis.IRedisPublisher;
import com.example.redis.impl.RedisMessageListener;
import com.example.redis.impl.RedisPublisherImpl;

@Configuration
@EnableScheduling
public class AppConfig {
    @Bean
    JedisConnectionFactory jedisConnectionFactory() {
        return new JedisConnectionFactory();
    }

    @Bean
    RedisTemplate< String, Object > redisTemplate() {
        final RedisTemplate< String, Object > template =  new RedisTemplate< String, Object >();
        template.setConnectionFactory( jedisConnectionFactory() );
        template.setKeySerializer( new StringRedisSerializer() );
        template.setHashValueSerializer( new GenericToStringSerializer< Object >( Object.class ) );
        template.setValueSerializer( new GenericToStringSerializer< Object >( Object.class ) );
        return template;
    }

    @Bean
    MessageListenerAdapter messageListener() {
        return new MessageListenerAdapter( new RedisMessageListener() );
    }

    @Bean
    RedisMessageListenerContainer redisContainer() {
        final RedisMessageListenerContainer container = new RedisMessageListenerContainer();

        container.setConnectionFactory( jedisConnectionFactory() );
        container.addMessageListener( messageListener(), topic() );

        return container;
    }
 
    @Bean
    IRedisPublisher redisPublisher() {
        return new RedisPublisherImpl( redisTemplate(), topic() );
    }

    @Bean
    ChannelTopic topic() {
        return new ChannelTopic( "pubsub:queue" );
    }
}

Very easy and straightforward. The presence of @EnableScheduling annotation is not necessary and is required only for our publisher implementation: the publisher will publish a string message every 100 ms.

package com.example.redis.impl;

import java.util.concurrent.atomic.AtomicLong;

import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.scheduling.annotation.Scheduled;

import com.example.redis.IRedisPublisher;

public class RedisPublisherImpl implements IRedisPublisher {
    private final RedisTemplate< String, Object > template;
    private final ChannelTopic topic; 
    private final AtomicLong counter = new AtomicLong( 0 );

    public RedisPublisherImpl( final RedisTemplate< String, Object > template, 
            final ChannelTopic topic ) {
        this.template = template;
        this.topic = topic;
    }

    @Scheduled( fixedDelay = 100 )
    public void publish() {
        template.convertAndSend( topic.getTopic(), "Message " + counter.incrementAndGet() + 
            ", " + Thread.currentThread().getName() );
 }
}

And finally our message listener implementation (which just prints message on a console).

package com.example.redis.impl;

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

public class RedisMessageListener implements MessageListener {
    @Override
    public void onMessage( final Message message, final byte[] pattern ) {
        System.out.println( "Message received: " + message.toString() );
    }
}

Awesome, just two small classes, one configuration to wire things together and we have full pub/sub messaging support in our application! Let's run the application as standalone ...

package com.example.redis;

import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import com.example.redis.config.AppConfig;

public class RedisPubSubStarter {
    public static void main(String[] args) {
        new AnnotationConfigApplicationContext( AppConfig.class );
    }
}
... and see following output in a console:
...
Message received: Message 1, pool-1-thread-1
Message received: Message 2, pool-1-thread-1
Message received: Message 3, pool-1-thread-1
Message received: Message 4, pool-1-thread-1
Message received: Message 5, pool-1-thread-1
Message received: Message 6, pool-1-thread-1
Message received: Message 7, pool-1-thread-1
Message received: Message 8, pool-1-thread-1
Message received: Message 9, pool-1-thread-1
Message received: Message 10, pool-1-thread-1
Message received: Message 11, pool-1-thread-1
Message received: Message 12, pool-1-thread-1
Message received: Message 13, pool-1-thread-1
Message received: Message 14, pool-1-thread-1
Message received: Message 15, pool-1-thread-1
Message received: Message 16, pool-1-thread-1
...
Great! There is much more which you could do with Redis pub/sub, excellent documentation is available for you on Redis official web site.

Monday, June 25, 2012

Using Redis with Spring

As NoSQL solutions are getting more and more popular for many kind of problems, more often the modern projects consider to use some (or several) of NoSQLs instead (or side-by-side) of traditional RDBMS. I have already covered my experience with MongoDB in this, this and this posts. In this post I would like to switch gears a bit towards Redis, an advanced key-value store.

Aside from very rich key-value semantics, Redis also supports pub-sub messaging and transactions. In this post I am going just to touch the surface and demonstrate how simple it is to integrate Redis into your Spring application.

As always, we will start with Maven POM file for our project:


 4.0.0
 com.example.spring
 redis
 0.0.1-SNAPSHOT
 jar

 
  UTF-8
  3.1.0.RELEASE
 

 
  
   org.springframework.data
   spring-data-redis
   1.0.0.RELEASE
  

  
   cglib
   cglib-nodep
   2.2
  

  
   log4j
   log4j
   1.2.16
  

  
   redis.clients
   jedis
   2.0.0
   jar
  

  
   org.springframework
   spring-core
   ${spring.version}
  

  
   org.springframework
   spring-context
   ${spring.version}
  
 

Spring Data Redis is the another project under Spring Data umbrella which provides seamless injection of Redis into your application. The are several Redis clients for Java and I have chosen the Jedis as it is stable and recommended by Redis team at the moment of writing this post.

We will start with simple configuration and introduce the necessary components first. Then as we move forward, the configuration will be extended a bit to demonstrated pub-sub capabilities. Thanks to Java config support, we will create the configuration class and have all our dependencies strongly typed, no XML anymore:

package com.example.redis.config;

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.GenericToStringSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;

@Configuration
public class AppConfig {
 @Bean
 JedisConnectionFactory jedisConnectionFactory() {
  return new JedisConnectionFactory();
 }

 @Bean
 RedisTemplate< String, Object > redisTemplate() {
  final RedisTemplate< String, Object > template =  new RedisTemplate< String, Object >();
  template.setConnectionFactory( jedisConnectionFactory() );
  template.setKeySerializer( new StringRedisSerializer() );
  template.setHashValueSerializer( new GenericToStringSerializer< Object >( Object.class ) );
  template.setValueSerializer( new GenericToStringSerializer< Object >( Object.class ) );
  return template;
 }
}
That's basically everything we need assuming we have single Redis server up and running on localhost with default configuration. Let's consider several common uses cases: setting a key to some value, storing the object and, finally, pub-sub implementation. Storing and retrieving a key/value pair is very simple:
@Autowired private RedisTemplate< String, Object > template;

public Object getValue( final String key ) {
    return template.opsForValue().get( key );
}

public void setValue( final String key, final String value ) {
    template.opsForValue().set( key, value );
}
Optionally, the key could be set to expire (yet another useful feature of Redis), f.e. let our keys expire in 1 second:
public void setValue( final String key, final String value ) {
    template.opsForValue().set( key, value );
    template.expire( key, 1, TimeUnit.SECONDS );
}
Arbitrary objects could be saved into Redis as hashes (maps), f.e. let save instance of some class User
public class User {
 private final Long id;
 private String name;
 private String email;
       
    // Setters and getters are omitted for simplicity
}
into Redis using key pattern "user:<id>":
public void setUser( final User user ) {
 final String key = String.format( "user:%s", user.getId() );
 final Map< String, Object > properties = new HashMap< String, Object >();

 properties.put( "id", user.getId() );
 properties.put( "name", user.getName() );
 properties.put( "email", user.getEmail() );

 template.opsForHash().putAll( key, properties);
}
Respectively, object could easily be inspected and retrieved using the id.
public User getUser( final Long id ) {
 final String key = String.format( "user:%s", id );

 final String name = ( String )template.opsForHash().get( key, "name" );
 final String email = ( String )template.opsForHash().get( key, "email" );

 return new User( id, name, email );
}
There are much, much more which could be done using Redis, I highly encourage to take a look on it. It surely is not a silver bullet but could solve many challenging problems very easy. Finally, let me show how to use a pub-sub messaging with Redis. Let's add a bit more configuration here (as part of AppConfig class):
@Bean
MessageListenerAdapter messageListener() {
 return new MessageListenerAdapter( new RedisMessageListener() );
}

@Bean
RedisMessageListenerContainer redisContainer() {
 final RedisMessageListenerContainer container = new RedisMessageListenerContainer();

 container.setConnectionFactory( jedisConnectionFactory() );
 container.addMessageListener( messageListener(), new ChannelTopic( "my-queue" ) );

 return container;
}
The style of message listener definition should look very familiar to Spring users: generally, the same approach we follow to define JMS message listeners. The missed piece is our RedisMessageListener class definition:
package com.example.redis.impl;

import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;

public class RedisMessageListener implements MessageListener {
 @Override
 public void onMessage(Message message, byte[] paramArrayOfByte) {
  System.out.println( "Received by RedisMessageListener: " + message.toString() );
 }
}
Now, when we have our message listener, let see how we could push some messages into the queue using Redis. As always, it's pretty simple:
@Autowired private RedisTemplate< String, Object > template;

public void publish( final String message ) {
 template.execute(
  new RedisCallback< Long >() {
   @SuppressWarnings( "unchecked" )
   @Override
   public Long doInRedis( RedisConnection connection ) throws DataAccessException {
    return connection.publish(
     ( ( RedisSerializer< String > )template.getKeySerializer() ).serialize( "queue" ),
     ( ( RedisSerializer< Object > )template.getValueSerializer() ).serialize( message ) );
   }
  }
 );
}
That's basically it for very quick introduction but definitely enough to fall in love with Redis.