diff --git a/README.Predis.txt b/README.Predis.txt
index 5f821df..32968a1 100644
--- a/README.Predis.txt
+++ b/README.Predis.txt
@@ -1,57 +1,45 @@
 Predis cache backend
 ====================
 
-This client, for now, is only able to use the Predis PHP library.
-
-The Predis library requires PHP 5.3 minimum. If your hosted environment does
-not ships with at least PHP 5.3, please do not use this cache backend.
-
-Please consider using an OPCode cache such as APC. Predis is a good and fully
-featured API, the cost is that the code is a lot more than a single file in
-opposition to some other backends such as the APC one.
+Using Predis for the Drupal 8 version of this module is still experimental.
 
 Get Predis
 ----------
 
-You can download this library at:
+Predis can be installed to the vendor directory using composer like so:
 
-  https://github.com/nrk/predis
+composer require drupal/redis
 
-This file explains how to install the Predis library and the Drupal cache
-backend. If you are an advanced Drupal integrator, please consider the fact
-that you can easily change all the pathes. Pathes used in this file are
-likely to be default for non advanced users.
+The library is listed as a dependency of this module and should be installed automatically.
 
-Download and install library
+Configuration of module for use with Predis
 ----------------------------
 
-Once done, you either have to clone it into:
-
-  sites/all/libraries/predis
-
-So that you have the following directory tree:
-
-  sites/all/libraries/lib/Predis # Where the PHP code stands
-
-Or, any other place in order to share it:
-For example, into your libraries folder, in order to get:
-
-  some/dir/predis/lib
-
-If you choose this solution, you have to alter a bit your $conf array into
-the settings.php file as this:
-
-  define('PREDIS_BASE_PATH', DRUPAL_ROOT . '/some/dir/predis/lib/');
-
-Connect to a remote host and database
--------------------------------------
-
-See README.txt file.
-
-Advanced configuration (PHP expert)
------------------------------------
+There is not much different to configure about Predis.
+Adding this to settings.php should suffice for basic usage:
+
+$settings['redis.connection']['interface'] = 'Predis';
+$settings['redis.connection']['host']      = '1.2.3.4';  // Your Redis instance hostname.
+$settings['cache']['default'] = 'cache.backend.redis';
+
+To add more magic with a primary/replica setup you can use a config like this:
+
+$settings['redis.connection']['interface'] = 'Predis'; // Use predis library.
+$settings['redis.connection']['replication'] = TRUE; // Turns on replication.
+$settings['redis.connection']['replication.host'][1]['host'] = '1.2.3.4';  // Your Redis instance hostname.
+$settings['redis.connection']['replication.host'][1]['port'] = '6379'; // Only required if using non-standard port.
+$settings['redis.connection']['replication.host'][1]['role'] = 'primary'; // The redis instance role.
+$settings['redis.connection']['replication.host'][2]['host'] = '1.2.3.5';
+$settings['redis.connection']['replication.host'][2]['port'] = '6379';
+$settings['redis.connection']['replication.host'][2]['role'] = 'replica';
+$settings['redis.connection']['replication.host'][3]['host'] = '1.2.3.6';
+$settings['redis.connection']['replication.host'][3]['port'] = '6379';
+$settings['redis.connection']['replication.host'][3]['role'] = 'replica';
+$settings['cache']['default'] = 'cache.backend.redis';
+
+Always set the fast backend for bootstrap, discover and config, otherwise
+this gets lost when redis is enabled.
+$settings['cache']['bins']['bootstrap'] = 'cache.backend.chainedfast';
+$settings['cache']['bins']['discovery'] = 'cache.backend.chainedfast';
+$settings['cache']['bins']['config'] = 'cache.backend.chainedfast';
 
-Best solution is, whatever is the place where you put the Predis library, that
-you set up a fully working autoloader able to use it. The one being used by the
-Redis module is a default fallback and will naturally being appened to the SPL
-autoloader stack.
diff --git a/README.md b/README.md
index cf6511f..4e4a9df 100644
--- a/README.md
+++ b/README.md
@@ -14,8 +14,9 @@ will need to compile the extension yourself.
 Predis
 ------
 
-Support for the Predis PHP library has not yet been ported to Drupal 8.
-
+Support for the Predis PHP library is experimental, but feel free to try it out. 
+You can install the required library using composer. Check out the README.Predis.txt file 
+for more information.
 
 Important notice
 ----------------
@@ -43,6 +44,27 @@ This method will allow Drupal to use Redis for all caches.
     $settings['cache']['bins']['discovery'] = 'cache.backend.chainedfast';
     $settings['cache']['bins']['config'] = 'cache.backend.chainedfast';
 
+To use some Predis goodness, including a redis primary/replica setup you can use a config like this.
+
+    $settings['redis.connection']['interface'] = 'Predis'; // Use predis library.
+    $settings['redis.connection']['replication'] = TRUE; // Turns on replication.
+    $settings['redis.connection']['replication.host'][1]['host'] = '1.2.3.4';  // Your Redis instance hostname.
+    $settings['redis.connection']['replication.host'][1]['port'] = '6379'; // Only required if using non-standard port.
+    $settings['redis.connection']['replication.host'][1]['role'] = 'primary'; // The redis instance role.
+    $settings['redis.connection']['replication.host'][2]['host'] = '1.2.3.5';
+    $settings['redis.connection']['replication.host'][2]['port'] = '6379';
+    $settings['redis.connection']['replication.host'][2]['role'] = 'replica';
+    $settings['redis.connection']['replication.host'][3]['host'] = '1.2.3.6';
+    $settings['redis.connection']['replication.host'][3]['port'] = '6379';
+    $settings['redis.connection']['replication.host'][3]['role'] = 'replica';
+    $settings['cache']['default'] = 'cache.backend.redis';
+
+    // Always set the fast backend for bootstrap, discover and config, otherwise
+    // this gets lost when redis is enabled.
+    $settings['cache']['bins']['bootstrap'] = 'cache.backend.chainedfast';
+    $settings['cache']['bins']['discovery'] = 'cache.backend.chainedfast';
+    $settings['cache']['bins']['config'] = 'cache.backend.chainedfast';
+
 Either include the default example.services.yml from the module, which will
 replace all supported backend services (that currently includes the cache tags
 checksum service and the lock backends, check the file for the current list)
diff --git a/composer.json b/composer.json
index 7ee969f..423d2f1 100644
--- a/composer.json
+++ b/composer.json
@@ -1,6 +1,9 @@
 {
   "name": "drupal/redis",
   "type": "drupal-module",
+  "require": {
+    "predis/predis": "^1.1.1"
+  },
   "license": "GPL-2.0",
   "autoload": {
     "psr-4": {
diff --git a/src/Cache/Predis.php b/src/Cache/Predis.php
index c62b05d..a2c303b 100644
--- a/src/Cache/Predis.php
+++ b/src/Cache/Predis.php
@@ -2,205 +2,269 @@
 
 namespace Drupal\redis\Cache;
 
+use Drupal\Component\Serialization\SerializationInterface;
 use Drupal\Core\Cache\Cache;
-use Drupal\redis\Cache\CacheBase;
+use Drupal\Core\Cache\CacheTagsChecksumInterface;
 
 /**
  * Predis cache backend.
  */
 class Predis extends CacheBase {
 
-  function get($cid, $allow_invalid = FALSE) {
-
-    $client = ClientFactory::getClient();
-    $key    = $this->getKey($cid);
-
-    $cached = $client->hgetall($key);
-
-    if (empty($cached)) {
-      return FALSE;
-    }
+  /**
+   * @var \Predis\Client
+   */
+  protected $client;
 
-    $cached = (object)$cached;
+  /**
+   * The cache tags checksum provider.
+   *
+   * @var \Drupal\Core\Cache\CacheTagsChecksumInterface|\Drupal\Core\Cache\CacheTagsInvalidatorInterface
+   */
+  protected $checksumProvider;
 
-    if ($cached->serialized) {
-      $cached->data = $this->serializer->decode($cached->data);
-    }
+  /**
+   * The last delete timestamp.
+   *
+   * @var float
+   */
+  protected $lastDeleteAll = NULL;
 
-    return $cached;
+  /**
+   * Creates a Predis cache backend.
+   *
+   * @param $bin
+   *   The cache bin for which the object is created.
+   * @param \Redis $client
+   * @param \Drupal\Core\Cache\CacheTagsChecksumInterface $checksum_provider
+   * @param \Drupal\redis\Cache\SerializationInterface $serializer
+   *   The serialization class to use.
+   */
+  public function __construct($bin, \Predis\Client $client, CacheTagsChecksumInterface $checksum_provider, SerializationInterface $serializer) {
+    parent::__construct($bin, $serializer);
+    $this->client = $client;
+    $this->checksumProvider = $checksum_provider;
   }
 
   /**
    * {@inheritdoc}
    */
-  function getMultiple(&$cids, $allow_invalid = FALSE) {
+  public function getMultiple(&$cids, $allow_invalid = FALSE) {
+    // Avoid an error when there are no cache ids.
+    if (empty($cids)) {
+      return [];
+    }
 
-    $client = ClientFactory::getClient();
-    $ret    = $keys = array();
-    $keys   = array_map(array($this, 'getKey'), $cids);
+    $return = array();
 
-    $replies = $client->pipeline(function($pipe) use ($keys) {
+    // Build the list of keys to fetch.
+    $keys = array_map(array($this, 'getKey'), $cids);
+
+    // Optimize for the common case when only a single cache entry needs to
+    // be fetched, no pipeline is needed then.
+    if (count($keys) > 1) {
+      $pipe = $this->client->pipeline();
       foreach ($keys as $key) {
         $pipe->hgetall($key);
       }
-    });
-
-    foreach ($replies as $reply) {
-      if (!empty($reply)) {
-
-        // HGETALL signature seems to differ depending on Predis versions.
-        // This was found just after Predis update. Even though I'm not sure
-        // this comes from Predis or just because we're misusing it.
-        // FIXME: Needs some investigation.
-        if (!isset($reply['cid'])) {
-          $cache = new stdClass();
-          $size = count($reply);
-          for ($i = 0; $i < $size; ++$i) {
-            $cache->{$reply[$i]} = $reply[++$i];
-          }
-        } else {
-          $cache = (object)$reply;
-        }
+      $result = $pipe->execute();
+    }
+    else {
+      $result = [$this->client->hGetAll(reset($keys))];
+    }
 
-        if ($cache->serialized) {
-          $cache->data = $this->serializer->decode($cache->data);
+    // Loop over the cid values to ensure numeric indexes.
+    foreach (array_values($cids) as $index => $key) {
+      // Check if a valid result was returned from Redis.
+      if (isset($result[$index]) && is_array($result[$index])) {
+        // Check expiration and invalidation and convert into an object.
+        $item = $this->expandEntry($result[$index], $allow_invalid);
+        if ($item) {
+          $return[$item->cid] = $item;
         }
-
-        $ret[$cache->cid] = $cache;
       }
     }
 
-    foreach ($cids as $index => $cid) {
-      if (isset($ret[$cid])) {
-        unset($cids[$index]);
-      }
-    }
+    // Remove fetched cids from the list.
+    $cids = array_diff($cids, array_keys($return));
 
-    return $ret;
+    return $return;
   }
 
   /**
    * {@inheritdoc}
    */
-  function set($cid, $data, $expire = Cache::PERMANENT, array $tags = array()) {
+  public function set($cid, $data, $expire = Cache::PERMANENT, array $tags = array()) {
 
-    $client = ClientFactory::getClient();
-    $skey   = $this->getKey(Redis_Cache_Base::TEMP_SET);
-    $key    = $this->getKey($cid);
-    $self   = $this;
+    $ttl = $this->getExpiration($expire);
 
-    $client->pipeline(function($pipe) use ($cid, $key, $skey, $data, $expire, $self) {
+    $key = $this->getKey($cid);
 
-      $hash = array(
-        'cid' => $cid,
-        'created' => time(),
-        'expire' => $expire,
-      );
+    // If the item is already expired, delete it.
+    if ($ttl <= 0) {
+      $this->delete($key);
+    }
 
-      if (!is_scalar($data)) {
-        $hash['data'] = $this->serializer->encode($data);
-        $hash['serialized'] = 1;
-      }
-      else {
-        $hash['data'] = $data;
-        $hash['serialized'] = 0;
-      }
+    // Build the cache item and save it as a hash array.
+    $entry = $this->createEntryHash($cid, $data, $expire, $tags);
+    $pipe = $this->client->pipeline();
+    $pipe->hmset($key, $entry);
+    $pipe->expire($key, $ttl);
+    $pipe->execute();
+  }
 
-      $pipe->hmset($key, $hash);
-
-      switch ($expire) {
-
-        case CACHE_TEMPORARY:
-          $lifetime = variable_get('cache_lifetime', Redis_Cache_Base::LIFETIME_DEFAULT);
-          if (0 < $lifetime) {
-            $pipe->expire($key, $lifetime);
-          }
-          $pipe->sadd($skey, $cid);
-          break;
-
-        case CACHE_PERMANENT:
-          if (0 !== ($ttl = $self->getPermTtl())) {
-            $pipe->expire($key, $ttl);
-          }
-          // We dont need the PERSIST command we want the cache item to
-          // never expire.
-          break;
-
-        default:
-          // If caller gives us an expiry timestamp in the past
-          // the key will expire now and will never be read.
-          $ttl = $expire - time();
-          $pipe->expire($key, $ttl);
-          if (0 < $ttl) {
-            $pipe->sadd($skey, $cid);
-          }
-          break;
-      }
-    });
+  /**
+   * {@inheritdoc}
+   */
+  public function deleteMultiple(array $cids) {
+    $keys = array_map(array($this, 'getKey'), $cids);
+    $this->client->del($keys);
   }
 
-  function clear($cid = NULL, $wildcard = FALSE) {
-
-    $keys   = array();
-    $skey   = $this->getKey(Redis_Cache_Base::TEMP_SET);
-    $client = ClientFactory::getClient();
-
-    if (NULL === $cid) {
-      switch ($this->getClearMode()) {
-
-        // One and only case of early return.
-        case Redis_Cache_Base::FLUSH_NOTHING:
-          return;
-
-        // Default behavior.
-        case Redis_Cache_Base::FLUSH_TEMPORARY:
-          if (Redis_Cache_Base::LIFETIME_INFINITE == variable_get('cache_lifetime', Redis_Cache_Base::LIFETIME_DEFAULT)) {
-            $keys[] = $skey;
-            foreach ($client->smembers($skey) as $tcid) {
-              $keys[] = $this->getKey($tcid);
-            }
-          }
-          break;
-
-        // Fallback on most secure mode: flush full bin.
-        default:
-        case Redis_Cache_Base::FLUSH_ALL:
-          $keys[] = $skey;
-          $cid = '*';
-          $wildcard = true;
-          break;
+  /**
+   * {@inheritdoc}
+   */
+  public function deleteAll() {
+    // The last delete timestamp is in milliseconds, ensure that no cache
+    // was written in the same millisecond.
+    // @todo This is needed to make the tests pass, is this safe enough for real
+    //   usage?
+    usleep(1000);
+    $this->lastDeleteAll = round(microtime(TRUE), 3);
+    $this->client->set($this->getKey(static::LAST_DELETE_ALL_KEY), $this->lastDeleteAll);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function invalidateMultiple(array $cids) {
+    // Loop over all cache items, they are stored as a hash, so we can access
+    // the valid flag directly, only write if it exists and is not 0.
+    foreach ($cids as $cid) {
+      $key = $this->getKey($cid);
+      if ($this->client->hGet($key, 'valid')) {
+        $this->client->hSet($key, 'valid', 0);
       }
     }
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function invalidateAll() {
+    // To invalidate the whole bin, we invalidate a special tag for this bin.
+    $this->checksumProvider->invalidateTags([$this->getTagForBin()]);
+  }
 
-    if ('*' !== $cid && $wildcard) {
-      // Prefix flush.
-      $keys = array_merge($keys, $client->keys($this->getKey($cid . '*')));
+  /**
+   * {@inheritdoc}
+   */
+  public function garbageCollection() {
+    // @todo Do we need to do anything here?
+  }
+
+  /**
+   *  Returns the last delete all timestamp.
+   *
+   * @return float
+   *   The last delete timestamp as a timestamp with a millisecond precision.
+   */
+  protected function getLastDeleteAll() {
+    // Cache the last delete all timestamp.
+    if ($this->lastDeleteAll === NULL) {
+      $this->lastDeleteAll = (float) $this->client->get($this->getKey(static::LAST_DELETE_ALL_KEY));
     }
-    else if ('*' === $cid) {
-      // Full bin flush.
-      $keys = array_merge($keys, $client->keys($this->getKey('*')));
+    return $this->lastDeleteAll;
+  }
+
+  /**
+   * Create cache entry.
+   *
+   * @param string $cid
+   * @param mixed $data
+   * @param int $expire
+   * @param string[] $tags
+   *
+   * @return array
+   */
+  protected function createEntryHash($cid, $data, $expire = Cache::PERMANENT, array $tags) {
+    // Always add a cache tag for the current bin, so that we can use that for
+    // invalidateAll().
+    $tags[] = $this->getTagForBin();
+    assert('\Drupal\Component\Assertion\Inspector::assertAllStrings($tags)', 'Cache Tags must be strings.');
+    $hash = array(
+      'cid' => $cid,
+      'created' => round(microtime(TRUE), 3),
+      'expire' => $expire,
+      'tags' => implode(' ', $tags),
+      'valid' => 1,
+      'checksum' => $this->checksumProvider->getCurrentChecksum($tags),
+    );
+
+    // Let Redis handle the data types itself.
+    if (!is_string($data)) {
+      $hash['data'] = $this->serializer->encode($data);
+      $hash['serialized'] = 1;
     }
-    else if (empty($keys) && !empty($cid)) {
-      // Single key drop.
-      $keys[] = $key = $this->getKey($cid);
-      $client->srem($skey, $key);
+    else {
+      $hash['data'] = $data;
+      $hash['serialized'] = 0;
     }
 
-    if (!empty($keys)) {
-      if (count($keys) < Redis_Cache_Base::KEY_THRESHOLD) {
-        $client->del($keys);
-      } else {
-        $client->pipeline(function($pipe) use ($keys) {
-          do {
-            $buffer = array_splice($keys, 0, Redis_Cache_Base::KEY_THRESHOLD);
-            $pipe->del($buffer);
-          } while (!empty($keys));
-        });
+    return $hash;
+  }
+
+  /**
+   * Prepares a cached item.
+   *
+   * Checks that items are either permanent or did not expire, and unserializes
+   * data as appropriate.
+   *
+   * @param array $values
+   *   The hash returned from redis or false.
+   * @param bool $allow_invalid
+   *   If FALSE, the method returns FALSE if the cache item is not valid.
+   *
+   * @return mixed|false
+   *   The item with data unserialized as appropriate and a property indicating
+   *   whether the item is valid, or FALSE if there is no valid item to load.
+   */
+  protected function expandEntry(array $values, $allow_invalid) {
+    // Check for entry being valid.
+    if (empty($values['cid'])) {
+      return FALSE;
+    }
+
+    $cache = (object) $values;
+
+    $cache->tags = explode(' ', $cache->tags);
+
+    // Check expire time, allow to have a cache invalidated explicitly, don't
+    // check if already invalid.
+    if ($cache->valid) {
+      $cache->valid = $cache->expire == Cache::PERMANENT || $cache->expire >= REQUEST_TIME;
+
+      // Check if invalidateTags() has been called with any of the items's tags.
+      if ($cache->valid && !$this->checksumProvider->isValid($cache->checksum, $cache->tags)) {
+        $cache->valid = FALSE;
       }
     }
-  }
 
-  function isEmpty() {
-    // FIXME: Todo.
+    // Ensure the entry does not predate the last delete all time.
+    $last_delete_timestamp = $this->getLastDeleteAll();
+    if ($last_delete_timestamp && ((float)$values['created']) < $last_delete_timestamp) {
+      return FALSE;
+    }
+
+    if (!$allow_invalid && !$cache->valid) {
+      return FALSE;
+    }
+
+    if ($cache->serialized) {
+      $cache->data = $this->serializer->decode($cache->data);
+    }
+
+    return $cache;
   }
+
 }
diff --git a/src/Cache/RedisCacheTagsChecksum.php b/src/Cache/RedisCacheTagsChecksum.php
index d41d157..b03c697 100644
--- a/src/Cache/RedisCacheTagsChecksum.php
+++ b/src/Cache/RedisCacheTagsChecksum.php
@@ -31,15 +31,21 @@ class RedisCacheTagsChecksum implements CacheTagsChecksumInterface, CacheTagsInv
   protected $invalidatedTags = array();
 
   /**
-   * @var \Redis
+   * {@inheritdoc}
    */
   protected $client;
 
   /**
+   * @var string
+   */
+  protected $clientType;
+
+  /**
    * Creates a PHpRedis cache backend.
    */
   public function __construct(ClientFactory $factory) {
     $this->client = $factory->getClient();
+    $this->clientType = $factory->getClientName();
   }
 
   /**
@@ -57,12 +63,25 @@ class RedisCacheTagsChecksum implements CacheTagsChecksumInterface, CacheTagsInv
       $keys_to_increment[] = $this->getTagKey($tag);
     }
     if ($keys_to_increment) {
-      $multi = $this->client->multi(\Redis::PIPELINE);
-      foreach ($keys_to_increment as $key) {
-        $multi->incr($key);
+
+      // We want to differentiate between PhpRedis and Redis clients.
+      if ($this->clientType === 'PhpRedis') {
+        $multi = $this->client->multi(\Redis::PIPELINE);
+        foreach ($keys_to_increment as $key) {
+          $multi->incr($key);
+        }
+        $multi->exec();
+      }
+      elseif ($this->clientType === 'Predis') {
+
+        $pipe = $this->client->pipeline();
+        foreach ($keys_to_increment as $key) {
+          $pipe->incr($key);
+        }
+        $pipe->execute();
       }
-      $multi->exec();
     }
+
   }
 
   /**
diff --git a/src/Client/Predis.php b/src/Client/Predis.php
index b9013c3..e20ea88 100644
--- a/src/Client/Predis.php
+++ b/src/Client/Predis.php
@@ -3,80 +3,21 @@
 namespace Drupal\redis\Client;
 
 use Drupal\redis\ClientInterface;
+use Predis\Client;
+
 
 /**
  * Predis client specific implementation.
  */
 class Predis implements ClientInterface {
 
-  /**
-   * Circular depedency breaker.
-   */
-  static protected $autoloaderRegistered = false;
-
-  /**
-   * Define Predis base path if not already set, and if we need to set the
-   * autoloader by ourself. This will ensure no crash. Best way would have
-   * been that Drupal ships a PSR-0 autoloader, in which we could manually
-   * add our library path.
-   *
-   * We cannot do that in the file header, PHP class_exists() function wont
-   * see classes being loaded during the autoloading because this file is
-   * loaded by another autoloader: attempting the class_exists() during a
-   * pending autoloading would cause PHP to crash and ignore the rest of the
-   * file silentely (WTF!?). By delaying this at the getClient() call we
-   * ensure we are not in the class loading process anymore.
-   */
-  public static function setPredisAutoload() {
-
-    if (self::$autoloaderRegistered) {
-      return;
-    } else {
-      self::$autoloaderRegistered = true;
-    }
-
-    // If you attempt to set Drupal's bin cache_bootstrap using Redis, you
-    // will experience an infinite loop (breaking by itself the second time
-    // it passes by): the following call will wake up autoloaders (and we
-    // want that to work since user may have set its own autoloader) but
-    // will wake up Drupal's one too, and because Drupal core caches its
-    // file map, this will trigger this method to be called a second time
-    // and boom! Adios bye bye. That's why this will be called early in the
-    // 'redis.autoload.inc' file instead.
-    if (!class_exists('Predis\Client')) {
-
-      if (!defined('PREDIS_BASE_PATH')) {
-        $search = DRUPAL_ROOT . '/sites/all/libraries/predis/lib/';
-        if (is_dir($search)) {
-          define('PREDIS_BASE_PATH', $search);
-        } else {
-          throw new Exception("PREDIS_BASE_PATH constant must be set, Predis library must live in sites/all/libraries/predis.");
-        }
-      }
-
-      if (class_exists('AutoloadEarly')) {
-        AutoloadEarly::getInstance()->registerNamespace('Predis', PREDIS_BASE_PATH);
-      } else {
-        // Register a simple autoloader for Predis library. Since the Predis
-        // library is PHP 5.3 only, we can afford doing closures safely.
-        spl_autoload_register(function($classname) {
-          if (0 === strpos($classname, 'Predis\\')) {
-            $filename = PREDIS_BASE_PATH . str_replace('\\', '/', $classname) . '.php';
-            return (bool)require_once $filename;
-          }
-          return false;
-        });
-      }
-    }
-  }
-
-  public function getClient($host = NULL, $port = NULL, $base = NULL, $password = NULL) {
-    $connectionInfo = array(
+  public function getClient($host = NULL, $port = NULL, $base = NULL, $password = NULL, $replicationHosts = NULL) {
+    $connectionInfo = [
       'password' => $password,
       'host'     => $host,
       'port'     => $port,
       'database' => $base
-    );
+    ];
 
     foreach ($connectionInfo as $key => $value) {
       if (!isset($value)) {
@@ -90,9 +31,28 @@ class Predis implements ClientInterface {
     // account has logged in.
     date_default_timezone_set(@date_default_timezone_get());
 
-    $client = new \Predis\Client($connectionInfo);
+    // If we are passed in an array of $replicationHosts, we should attempt a clustered client connection.
+    if ($replicationHosts !== NULL) {
+      $parameters = [];
+
+      foreach ($replicationHosts as $replicationHost) {
+        // Configure master.
+        if ($replicationHost['role'] === 'primary') {
+          $parameters[] = 'tcp://' . $replicationHost['host'] . ':' . $replicationHost['port'] . '?alias=master';
+        }
+        else {
+          $parameters[] = 'tcp://' . $replicationHost['host'] . ':' . $replicationHost['port'];
+        }
+      }
 
+      $options = array('replication' => true);
+      $client = new Client($parameters, $options);
+    }
+    else {
+      $client = new Client($connectionInfo);
+    }
     return $client;
+
   }
 
   public function getName() {
diff --git a/src/ClientFactory.php b/src/ClientFactory.php
index 2cd07b7..28cd76f 100644
--- a/src/ClientFactory.php
+++ b/src/ClientFactory.php
@@ -152,12 +152,27 @@ class ClientFactory {
         'password' => self::REDIS_DEFAULT_PASSWORD,
       );
 
-      // Always prefer socket connection.
-      self::$_client = self::getClientInterface()->getClient(
-        $settings['host'],
-        $settings['port'],
-        $settings['base'],
-        $settings['password']);
+      foreach ($settings['replication.host'] as $key => $replicationHost) {
+        if (!isset($replicationHost['port'])) {
+          $settings['replication.host'][$key]['port'] = self::REDIS_DEFAULT_PORT;
+        }
+      }
+
+      if ($settings['replication'] === TRUE) {
+        self::$_client = self::getClientInterface()->getClient(
+          $settings['host'],
+          $settings['port'],
+          $settings['base'],
+          $settings['password'],
+          $settings['replication.host']);
+      }
+      else {
+        self::$_client = self::getClientInterface()->getClient(
+          $settings['host'],
+          $settings['port'],
+          $settings['base'],
+          $settings['password']);
+      }
     }
 
     return self::$_client;
diff --git a/src/Flood/Predis.php b/src/Flood/Predis.php
new file mode 100644
index 0000000..de28b04
--- /dev/null
+++ b/src/Flood/Predis.php
@@ -0,0 +1,95 @@
+<?php
+
+namespace Drupal\redis\Flood;
+
+use Drupal\Core\Flood\FloodInterface;
+use Drupal\redis\ClientFactory;
+use Drupal\redis\RedisPrefixTrait;
+use Symfony\Component\HttpFoundation\RequestStack;
+
+/**
+ * Defines the database flood backend. This is the default Drupal backend.
+ */
+class Predis implements FloodInterface {
+
+  use RedisPrefixTrait;
+
+  /**
+   * @var \Predis\Client
+   */
+  protected $client;
+
+  /**
+   * The request stack.
+   *
+   * @var \Symfony\Component\HttpFoundation\RequestStack
+   */
+  protected $requestStack;
+
+  /**
+   * Construct the PhpRedis flood backend.
+   *
+   * @param \Drupal\redis\ClientFactory $client_factory
+   *   The database connection which will be used to store the flood event
+   *   information.
+   * @param \Symfony\Component\HttpFoundation\RequestStack $request_stack
+   *   The request stack used to retrieve the current request.
+   */
+  public function __construct(ClientFactory $client_factory, RequestStack $request_stack) {
+    $this->client = $client_factory->getClient();
+    $this->requestStack = $request_stack;
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function register($name, $window = 3600, $identifier = NULL) {
+    if (!isset($identifier)) {
+      $identifier = $this->requestStack->getCurrentRequest()->getClientIp();
+    }
+
+    $key = $this->getPrefix() . ':flood:' . $name . ':' . $identifier;
+
+    // Add a key for the event to the sorted set, the score is timestamp, so we
+    // can count them easily.
+    $this->client->zAdd($key, $_SERVER['REQUEST_TIME'] + $window, microtime(TRUE));
+    // Set or update the expiration for the sorted set, it will be removed if
+    // the newest entry expired.
+    $this->client->expire($key, $_SERVER['REQUEST_TIME'] + $window);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function clear($name, $identifier = NULL) {
+    if (!isset($identifier)) {
+      $identifier = $this->requestStack->getCurrentRequest()->getClientIp();
+    }
+
+    $key = $this->getPrefix() . ':flood:' . $name . ':' . $identifier;
+    $this->client->del($key);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function isAllowed($name, $threshold, $window = 3600, $identifier = NULL) {
+    if (!isset($identifier)) {
+      $identifier = $this->requestStack->getCurrentRequest()->getClientIp();
+    }
+
+    $key = $this->getPrefix() . ':flood:' . $name . ':' . $identifier;
+
+    // Count the in the last $window seconds.
+    $number = $this->client->zCount($key, $_SERVER['REQUEST_TIME'], 'inf');
+    return ($number < $threshold);
+  }
+
+  /**
+   * {@inheritdoc}
+   */
+  public function garbageCollection() {
+    // No garbage collection necessary.
+  }
+
+}
diff --git a/src/Lock/Predis.php b/src/Lock/Predis.php
index 0a507c2..0f31181 100644
--- a/src/Lock/Predis.php
+++ b/src/Lock/Predis.php
@@ -3,136 +3,133 @@
 namespace Drupal\redis\Lock;
 
 use Drupal\Core\Lock\LockBackendAbstract;
-use Drupal\redis\LockBase;
+use Drupal\redis\ClientFactory;
+use Drupal\redis\RedisPrefixTrait;
 
 /**
  * Predis lock backend implementation.
  */
 class Predis extends LockBackendAbstract {
 
-  public function lockAcquire($name, $timeout = 30.0) {
-    $client = ClientFactory::getClient();
+  use RedisPrefixTrait;
+
+  /**
+   * @var \Predis\Client
+   */
+  protected $client;
+
+  /**
+   * Creates a PHpRedis cache backend.
+   */
+  public function __construct(ClientFactory $factory) {
+    $this->client = $factory->getClient();
+    // __destruct() is causing problems with garbage collections, register a
+    // shutdown function instead.
+    drupal_register_shutdown_function(array($this, 'releaseAll'));
+  }
+
+  /**
+   * Generate a redis key name for the current lock name.
+   *
+   * @param string $name
+   *   Lock name.
+   *
+   * @return string
+   *   The redis key for the given lock.
+   */
+  protected function getKey($name) {
+    return $this->getPrefix() . ':lock:' . $name;
+  }
+
+  public function acquire($name, $timeout = 30.0) {
     $key    = $this->getKey($name);
     $id     = $this->getLockId();
 
-    // Insure that the timeout is at least 1 second, we cannot do otherwise with
-    // Redis, this is a minor change to the function signature, but in real life
-    // nobody will notice with so short duration.
-    $timeout = ceil(max($timeout, 1));
+    // Insure that the timeout is at least 1 ms.
+    $timeout = max($timeout, 0.001);
 
     // If we already have the lock, check for his owner and attempt a new EXPIRE
     // command on it.
-    if (isset($this->_locks[$name])) {
+    if (isset($this->locks[$name])) {
 
       // Create a new transaction, for atomicity.
-      $client->watch($key);
+      $this->client->watch($key);
 
       // Global tells us we are the owner, but in real life it could have expired
       // and another process could have taken it, check that.
-      if ($client->get($key) != $id) {
-        $client->unwatch($key);
-        unset($this->_locks[$name]);
+      if ($this->client->get($key) != $id) {
+        // Explicit UNWATCH we are not going to run the MULTI/EXEC block.
+        $this->client->unwatch();
+        unset($this->locks[$name]);
         return FALSE;
       }
 
-      $replies = $client->pipeline(function($pipe) use ($key, $timeout, $id) {
-        $pipe->multi();
-        $pipe->setex($key, $timeout, $id);
-        $pipe->exec();
-      });
-
-      $execReply = array_pop($replies);
+      $result = $this->client->pipeline()
+        ->psetex($key, (int) ($timeout * 1000), $id)
+        ->exec();
 
-      if (FALSE === $execReply[0]) {
-        unset($this->_locks[$name]);
+      // If the set failed, someone else wrote the key, we failed to acquire
+      // the lock.
+      if (FALSE === $result) {
+        unset($this->locks[$name]);
+        // Explicit transaction release which also frees the WATCH'ed key.
+        $this->client->discard();
         return FALSE;
       }
 
-      return TRUE;
+      return ($this->locks[$name] = TRUE);
     }
     else {
-      $client->watch($key);
-      $owner = $client->get($key);
-
-      if (!empty($owner) && $owner != $id) {
-        $client->unwatch();
-        unset($this->_locks[$name]);
-        return FALSE;
-      }
-
-      $replies = $client->pipeline(function($pipe) use ($key, $timeout, $id) {
-        $pipe->multi();
-        $pipe->setex($key, $timeout, $id);
-        $pipe->exec();
-      });
-
-      $execReply = array_pop($replies);
+      // Use a SET with microsecond expiration and the NX flag, which will only
+      // succeed if the key does not exist yet.
+      $result = $this->client->set($key, $id, 'nx', 'px', (int) ($timeout * 1000));
 
-      // If another client modified the $key value, transaction will be discarded
-      // $result will be set to FALSE. This means atomicity have been broken and
-      // the other client took the lock instead of us.
-      // EXPIRE and SETEX won't return something here, EXEC return is index 0
-      // This was determined debugging, seems to be Predis specific.
-      if (FALSE === $execReply[0]) {
+      // If the result is FALSE, we failed to acquire the lock.
+      if (FALSE === $result) {
         return FALSE;
       }
 
-      // Register the lock and return.
-      return ($this->_locks[$name] = TRUE);
+      // Register the lock.
+      return ($this->locks[$name] = TRUE);
     }
-
-    return FALSE;
   }
 
   public function lockMayBeAvailable($name) {
-    $client = ClientFactory::getClient();
-    $key    = $this->getKey($name);
-    $id     = $this->getLockId();
+    $key = $this->getKey($name);
+    $value = $this->client->get($key);
 
-    $value = $client->get($key);
-
-    return empty($value) || $id == $value;
+    // In Drupal 7, this method treated the lock as available if the ID did
+    // match. The database backend and test expects it to return FALSE in that
+    // case, updated accordingly.
+    return FALSE === $value;
   }
 
-  public function lockRelease($name) {
-    $client = ClientFactory::getClient();
+  public function release($name) {
     $key    = $this->getKey($name);
     $id     = $this->getLockId();
 
-    unset($this->_locks[$name]);
+    unset($this->locks[$name]);
 
     // Ensure the lock deletion is an atomic transaction. If another thread
     // manages to removes all lock, we can not alter it anymore else we will
     // release the lock for the other thread and cause race conditions.
-    $client->watch($key);
+    $this->client->watch($key);
 
-    if ($client->get($key) == $id) {
-      $client->multi();
-      $client->del(array($key));
-      $client->exec();
+    if ($this->client->get($key) == $id) {
+      $pipe = $this->client->pipeline();
+      $pipe->del([$key]);
+      $pipe->execute();
     }
     else {
-      $client->unwatch();
+      $this->client->unwatch();
     }
   }
 
-  public function lockReleaseAll($lock_id = NULL) {
-    if (!isset($lock_id) && empty($this->_locks)) {
-      return;
-    }
-
-    $client = ClientFactory::getClient();
-    $id     = isset($lock_id) ? $lock_id : $this->getLockId();
-
+  public function releaseAll($lock_id = NULL) {
     // We can afford to deal with a slow algorithm here, this should not happen
     // on normal run because we should have removed manually all our locks.
-    foreach ($this->_locks as $name => $foo) {
-      $key   = $this->getKey($name);
-      $owner = $client->get($key);
-
-      if (empty($owner) || $owner == $id) {
-        $client->del(array($key));
-      }
+    foreach ($this->locks as $name => $foo) {
+      $this->release($name);
     }
   }
 }
diff --git a/src/Queue/ReliablePredis.php b/src/Queue/ReliablePredis.php
index 3a5043b..5d63be6 100644
--- a/src/Queue/ReliablePredis.php
+++ b/src/Queue/ReliablePredis.php
@@ -43,7 +43,7 @@ class ReliablePredis extends ReliableQueueBase {
     // by a single request which takes longer than 1 second.
     $record->timestamp = time();
 
-    $result = $this->client->multi()
+    $result = $this->client->pipeline()
       ->hsetnx($this->availableItems, $record->qid, serialize($record))
       ->lLen($this->availableListKey)
       ->lpush($this->availableListKey, $record->qid)
@@ -96,7 +96,7 @@ class ReliablePredis extends ReliableQueueBase {
    */
   public function releaseItem($item) {
     // TODO: Fixme
-    $this->client->multi()
+    $this->client->pipeline()
       ->lrem($this->claimedListKey, $item->qid, -1)
       ->lpush($this->availableListKey, $item->qid)
       ->exec();
@@ -107,7 +107,7 @@ class ReliablePredis extends ReliableQueueBase {
    */
   public function deleteItem($item) {
     // TODO: Fixme
-    $this->client->multi()
+    $this->client->pipeline()
       ->lrem($this->claimedListKey, $item->qid, -1)
       ->hdel($this->availableItems, $item->qid)
       ->exec();