diff --git a/modules/benchmarks/src/main/scala/scalacache/benchmark/CaffeineBenchmark.scala b/modules/benchmarks/src/main/scala/scalacache/benchmark/CaffeineBenchmark.scala
index 6ca5203d..5a99f1d5 100644
--- a/modules/benchmarks/src/main/scala/scalacache/benchmark/CaffeineBenchmark.scala
+++ b/modules/benchmarks/src/main/scala/scalacache/benchmark/CaffeineBenchmark.scala
@@ -17,8 +17,9 @@ class CaffeineBenchmark {
 
   implicit val clockSyncIO: Clock[SyncIO] = Clock[SyncIO]
 
-  val underlyingCache                       = Caffeine.newBuilder().build[String, Entry[String]]()
-  implicit val cache: Cache[SyncIO, String] = CaffeineCache[SyncIO, String](underlyingCache)
+  val underlyingCache = Caffeine.newBuilder().build[String, Entry[String]]()
+  implicit val cache: Cache[SyncIO, String, String] =
+    CaffeineCache[SyncIO, String, String](underlyingCache)
 
   val key           = "key"
   val value: String = "value"
diff --git a/modules/benchmarks/src/test/scala/scalacache/benchmark/ProfilingMemoize.scala b/modules/benchmarks/src/test/scala/scalacache/benchmark/ProfilingMemoize.scala
index ad94ba16..7be1add5 100644
--- a/modules/benchmarks/src/test/scala/scalacache/benchmark/ProfilingMemoize.scala
+++ b/modules/benchmarks/src/test/scala/scalacache/benchmark/ProfilingMemoize.scala
@@ -14,7 +14,7 @@ object ProfilingMemoize extends App {
 
   implicit val clockSyncIO = Clock[SyncIO]
   val underlyingCache      = Caffeine.newBuilder().build[String, Entry[String]]()
-  implicit val cache       = CaffeineCache[SyncIO, String](underlyingCache)
+  implicit val cache       = CaffeineCache[SyncIO, String, String](underlyingCache)
 
   val key           = "key"
   val value: String = "value"
diff --git a/modules/caffeine/src/main/scala/scalacache/caffeine/CaffeineCache.scala b/modules/caffeine/src/main/scala/scalacache/caffeine/CaffeineCache.scala
index 0155d641..45d40ee4 100644
--- a/modules/caffeine/src/main/scala/scalacache/caffeine/CaffeineCache.scala
+++ b/modules/caffeine/src/main/scala/scalacache/caffeine/CaffeineCache.scala
@@ -1,33 +1,28 @@
 package scalacache.caffeine
 
-import java.time.temporal.ChronoUnit
-import java.time.{Instant}
-
+import cats.effect.{Clock, Sync}
+import cats.implicits._
 import com.github.benmanes.caffeine.cache.{Caffeine, Cache => CCache}
-import cats.effect.Clock
 import scalacache.logging.Logger
-import scalacache.{AbstractCache, CacheConfig, Entry}
+import scalacache.{AbstractCache, Entry}
+
+import java.time.Instant
 import scala.concurrent.duration.Duration
 import scala.language.higherKinds
-import cats.effect.Sync
-import java.util.concurrent.TimeUnit
-import cats.implicits._
-import cats.MonadError
 
 /*
  * Thin wrapper around Caffeine.
  *
  * This cache implementation is synchronous.
  */
-class CaffeineCache[F[_]: Sync, V](val underlying: CCache[String, Entry[V]])(implicit
-    val config: CacheConfig,
-    clock: Clock[F]
-) extends AbstractCache[F, V] {
+class CaffeineCache[F[_]: Sync, K, V](val underlying: CCache[K, Entry[V]])(implicit
+    val clock: Clock[F]
+) extends AbstractCache[F, K, V] {
   protected val F: Sync[F] = Sync[F]
 
   override protected final val logger = Logger.getLogger(getClass.getName)
 
-  def doGet(key: String): F[Option[V]] = {
+  def doGet(key: K): F[Option[V]] = {
     F.delay {
       Option(underlying.getIfPresent(key))
     }.flatMap(_.filterA(Entry.isBeforeExpiration[F, V]))
@@ -37,7 +32,7 @@ class CaffeineCache[F[_]: Sync, V](val underlying: CCache[String, Entry[V]])(imp
       }
   }
 
-  def doPut(key: String, value: V, ttl: Option[Duration]): F[Unit] =
+  def doPut(key: K, value: V, ttl: Option[Duration]): F[Unit] =
     ttl.traverse(toExpiryTime).flatMap { expiry =>
       F.delay {
         val entry = Entry(value, expiry)
@@ -45,7 +40,7 @@ class CaffeineCache[F[_]: Sync, V](val underlying: CCache[String, Entry[V]])(imp
       } *> logCachePut(key, ttl)
     }
 
-  override def doRemove(key: String): F[Unit] =
+  override def doRemove(key: K): F[Unit] =
     F.delay(underlying.invalidate(key))
 
   override def doRemoveAll: F[Unit] =
@@ -65,17 +60,17 @@ object CaffeineCache {
 
   /** Create a new Caffeine cache.
     */
-  def apply[F[_]: Sync: Clock, V](implicit config: CacheConfig): F[CaffeineCache[F, V]] =
-    Sync[F].delay(Caffeine.newBuilder().build[String, Entry[V]]()).map(apply(_))
+  def apply[F[_]: Sync: Clock, K <: AnyRef, V]: F[CaffeineCache[F, K, V]] =
+    Sync[F].delay(Caffeine.newBuilder.build[K, Entry[V]]()).map(apply(_))
 
   /** Create a new cache utilizing the given underlying Caffeine cache.
     *
     * @param underlying
     *   a Caffeine cache
     */
-  def apply[F[_]: Sync: Clock, V](
-      underlying: CCache[String, Entry[V]]
-  )(implicit config: CacheConfig): CaffeineCache[F, V] =
+  def apply[F[_]: Sync: Clock, K, V](
+      underlying: CCache[K, Entry[V]]
+  ): CaffeineCache[F, K, V] =
     new CaffeineCache(underlying)
 
 }
diff --git a/modules/caffeine/src/test/scala/scalacache/caffeine/CaffeineCacheSpec.scala b/modules/caffeine/src/test/scala/scalacache/caffeine/CaffeineCacheSpec.scala
index c17cfd87..0e04064d 100644
--- a/modules/caffeine/src/test/scala/scalacache/caffeine/CaffeineCacheSpec.scala
+++ b/modules/caffeine/src/test/scala/scalacache/caffeine/CaffeineCacheSpec.scala
@@ -30,16 +30,18 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
     unsafeRun(f(ticker)) shouldBe Outcome.succeeded(Some(succeed))
   }
 
-  private def newCCache = Caffeine.newBuilder.build[String, Entry[String]]
+  case class MyInt(int: Int)
+
+  private def newCCache = Caffeine.newBuilder.build[MyInt, Entry[String]]
 
   private def newFCache[F[_]: Sync, V](
-      underlying: com.github.benmanes.caffeine.cache.Cache[String, Entry[V]]
+      underlying: com.github.benmanes.caffeine.cache.Cache[MyInt, Entry[V]]
   ) = {
-    CaffeineCache[F, V](underlying)
+    CaffeineCache[F, MyInt, V](underlying)
   }
 
   private def newIOCache[V](
-      underlying: com.github.benmanes.caffeine.cache.Cache[String, Entry[V]]
+      underlying: com.github.benmanes.caffeine.cache.Cache[MyInt, Entry[V]]
   ) = {
     newFCache[IO, V](underlying)
   }
@@ -49,14 +51,14 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
   it should "return the value stored in the underlying cache if expiration is not specified" in ticked { _ =>
     val underlying = newCCache
     val entry      = Entry("hello", expiresAt = None)
-    underlying.put("key1", entry)
+    underlying.put(MyInt(1), entry)
 
-    newIOCache(underlying).get("key1").map(_ shouldBe Some("hello"))
+    newIOCache(underlying).get(MyInt(1)).map(_ shouldBe Some("hello"))
   }
 
   it should "return None if the given key does not exist in the underlying cache" in ticked { _ =>
     val underlying = newCCache
-    newIOCache(underlying).get("non-existent key").map(_ shouldBe None)
+    newIOCache(underlying).get(MyInt(2)).map(_ shouldBe None)
   }
 
   it should "return None if the given key exists but the value has expired" in ticked { ticker =>
@@ -65,8 +67,8 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
       val underlying = newCCache
       val expiredEntry =
         Entry("hello", expiresAt = Some(Instant.ofEpochMilli(now.toMillis).minusSeconds(60)))
-      underlying.put("key1", expiredEntry)
-      newIOCache(underlying).get("key1").map(_ shouldBe None)
+      underlying.put(MyInt(1), expiredEntry)
+      newIOCache(underlying).get(MyInt(1)).map(_ shouldBe None)
     }
   }
 
@@ -76,8 +78,8 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
       val underlying = newCCache
       val expiredEntry =
         Entry("hello", expiresAt = Some(Instant.ofEpochMilli(now.toMillis).plusSeconds(60)))
-      underlying.put("key1", expiredEntry)
-      newIOCache(underlying).get("key1").map(_ shouldBe Some("hello"))
+      underlying.put(MyInt(1), expiredEntry)
+      newIOCache(underlying).get(MyInt(1)).map(_ shouldBe Some("hello"))
     }
   }
 
@@ -85,8 +87,8 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
 
   it should "store the given key-value pair in the underlying cache with no TTL" in ticked { _ =>
     val underlying = newCCache
-    newIOCache(underlying).put("key1")("hello", None) *>
-      IO { underlying.getIfPresent("key1") }
+    newIOCache(underlying).put(MyInt(1))("hello", None) *>
+      IO { underlying.getIfPresent(MyInt(1)) }
         .map(_ shouldBe Entry("hello", None))
   }
 
@@ -98,8 +100,8 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
 
     val underlying = newCCache
 
-    newFCache[IO, String](underlying).put("key1")("hello", Some(10.seconds)).map { _ =>
-      underlying.getIfPresent("key1") should be(Entry("hello", expiresAt = Some(now.plusSeconds(10))))
+    newFCache[IO, String](underlying).put(MyInt(1))("hello", Some(10.seconds)).map { _ =>
+      underlying.getIfPresent(MyInt(1)) should be(Entry("hello", expiresAt = Some(now.plusSeconds(10))))
     }
   }
 
@@ -108,8 +110,8 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
     val now = Instant.ofEpochMilli(ctx.now().toMillis)
 
     val underlying = newCCache
-    newFCache[IO, String](underlying).put("key1")("hello", Some(30.days)).map { _ =>
-      underlying.getIfPresent("key1") should be(
+    newFCache[IO, String](underlying).put(MyInt(1))("hello", Some(30.days)).map { _ =>
+      underlying.getIfPresent(MyInt(1)) should be(
         Entry("hello", expiresAt = Some(now.plusMillis(30.days.toMillis)))
       )
     }
@@ -120,11 +122,11 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
   it should "delete the given key and its value from the underlying cache" in ticked { _ =>
     val underlying = newCCache
     val entry      = Entry("hello", expiresAt = None)
-    underlying.put("key1", entry)
-    underlying.getIfPresent("key1") should be(entry)
+    underlying.put(MyInt(1), entry)
+    underlying.getIfPresent(MyInt(1)) should be(entry)
 
-    newIOCache(underlying).remove("key1") *>
-      IO(underlying.getIfPresent("key1")).map(_ shouldBe null)
+    newIOCache(underlying).remove(MyInt(1)) *>
+      IO(underlying.getIfPresent(MyInt(1))).map(_ shouldBe null)
   }
 
   behavior of "get after put"
@@ -132,8 +134,8 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
   it should "store the given key-value pair in the underlying cache with no TTL, then get it back" in ticked { _ =>
     val underlying = newCCache
     val cache      = newIOCache(underlying)
-    cache.put("key1")("hello", None) *>
-      cache.get("key1").map { _ shouldBe defined }
+    cache.put(MyInt(1))("hello", None) *>
+      cache.get(MyInt(1)).map { _ shouldBe defined }
   }
 
   behavior of "get after put with TTL"
@@ -143,8 +145,8 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
       val underlying = newCCache
       val cache      = newFCache[IO, String](underlying)
 
-      cache.put("key1")("hello", Some(5.seconds)) *>
-        cache.get("key1").map { _ shouldBe defined }
+      cache.put(MyInt(1))("hello", Some(5.seconds)) *>
+        cache.get(MyInt(1)).map { _ shouldBe defined }
   }
 
   it should "store the given key-value pair with the given TTL, then get it back (after a sleep) when not expired" in ticked {
@@ -152,9 +154,9 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
       val underlying = newCCache
       val cache      = newFCache[IO, String](underlying)
 
-      cache.put("key1")("hello", Some(50.seconds)) *>
+      cache.put(MyInt(1))("hello", Some(50.seconds)) *>
         IO.sleep(40.seconds) *> // sleep, but not long enough for the entry to expire
-        cache.get("key1").map { _ shouldBe defined }
+        cache.get(MyInt(1)).map { _ shouldBe defined }
   }
 
   it should "store the given key-value pair with the given TTL, then return None if the entry has expired" in ticked {
@@ -162,9 +164,9 @@ class CaffeineCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter wi
       val underlying = newCCache
       val cache      = newFCache[IO, String](underlying)
 
-      cache.put("key1")("hello", Some(50.seconds)) *>
+      cache.put(MyInt(1))("hello", Some(50.seconds)) *>
         IO.sleep(60.seconds) *> // sleep long enough for the entry to expire
-        cache.get("key1").map { _ shouldBe empty }
+        cache.get(MyInt(1)).map { _ shouldBe empty }
   }
 
 }
diff --git a/modules/circe/src/main/scala/scalacache/serialization/Circe.scala b/modules/circe/src/main/scala/scalacache/serialization/Circe.scala
index f8b898a1..ec62c117 100644
--- a/modules/circe/src/main/scala/scalacache/serialization/Circe.scala
+++ b/modules/circe/src/main/scala/scalacache/serialization/Circe.scala
@@ -1,21 +1,21 @@
 package scalacache.serialization
 
 import java.nio.ByteBuffer
-
 import io.circe.jawn.JawnParser
-import io.circe.{Decoder, Encoder}
+import scalacache.serialization.binary.BinaryCodec
 
 package object circe {
 
   private[this] val parser = new JawnParser
 
-  implicit def codec[A](implicit encoder: Encoder[A], decoder: Decoder[A]): Codec[A] = new Codec[A] {
+  implicit def codec[A](implicit encoder: io.circe.Encoder[A], decoder: io.circe.Decoder[A]): BinaryCodec[A] =
+    new BinaryCodec[A] {
 
-    override def encode(value: A): Array[Byte] = encoder.apply(value).noSpaces.getBytes("utf-8")
+      override def encode(value: A): Array[Byte] = encoder.apply(value).noSpaces.getBytes("utf-8")
 
-    override def decode(bytes: Array[Byte]): Codec.DecodingResult[A] =
-      parser.decodeByteBuffer(ByteBuffer.wrap(bytes)).left.map(FailedToDecode)
+      override def decode(bytes: Array[Byte]): Codec.DecodingResult[A] =
+        parser.decodeByteBuffer(ByteBuffer.wrap(bytes)).left.map(FailedToDecode)
 
-  }
+    }
 
 }
diff --git a/modules/circe/src/test/scala/scalacache/serialization/CirceCodecSpec.scala b/modules/circe/src/test/scala/scalacache/serialization/CirceCodecSpec.scala
index 0b5b87db..e07c5d2c 100644
--- a/modules/circe/src/test/scala/scalacache/serialization/CirceCodecSpec.scala
+++ b/modules/circe/src/test/scala/scalacache/serialization/CirceCodecSpec.scala
@@ -6,6 +6,7 @@ import io.circe.syntax._
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
 import org.scalatestplus.scalacheck.ScalaCheckDrivenPropertyChecks
+import scalacache.serialization.binary.BinaryCodec
 
 case class Fruit(name: String, tastinessQuotient: Double)
 
@@ -15,7 +16,7 @@ class CirceCodecSpec extends AnyFlatSpec with Matchers with ScalaCheckDrivenProp
 
   import scalacache.serialization.circe._
 
-  private def serdesCheck[A: Arbitrary](expectedJson: A => String)(implicit codec: Codec[A]): Unit = {
+  private def serdesCheck[A: Arbitrary](expectedJson: A => String)(implicit codec: BinaryCodec[A]): Unit = {
     forAll(minSuccessful(10000)) { (a: A) =>
       val serialised = codec.encode(a)
       new String(serialised, "utf-8") shouldBe expectedJson(a)
@@ -62,7 +63,7 @@ class CirceCodecSpec extends AnyFlatSpec with Matchers with ScalaCheckDrivenProp
 
   it should "serialize and deserialize a case class" in {
     import io.circe.generic.auto._
-    val fruitCodec = implicitly[Codec[Fruit]]
+    val fruitCodec = implicitly[BinaryCodec[Fruit]]
 
     val banana     = Fruit("banana", 0.7)
     val serialised = fruitCodec.encode(banana)
diff --git a/modules/core/src/main/scala-2/scalacache/memoization/Macros.scala b/modules/core/src/main/scala-2/scalacache/memoization/Macros.scala
index 8104dd6b..7051e532 100644
--- a/modules/core/src/main/scala-2/scalacache/memoization/Macros.scala
+++ b/modules/core/src/main/scala-2/scalacache/memoization/Macros.scala
@@ -1,38 +1,39 @@
 package scalacache.memoization
 
-import scala.language.experimental.macros
-import scala.reflect.macros.blackbox
+import scalacache.{Cache, Flags}
+
 import scala.concurrent.duration.Duration
+import scala.language.experimental.macros
 import scala.language.higherKinds
-import scalacache.{Flags, Cache}
+import scala.reflect.macros.blackbox
 
 class Macros(val c: blackbox.Context) {
   import c.universe._
 
   def memoizeImpl[F[_], V: c.WeakTypeTag](
       ttl: c.Expr[Option[Duration]]
-  )(f: c.Tree)(cache: c.Expr[Cache[F, V]], flags: c.Expr[Flags]): c.Tree = {
+  )(f: c.Tree)(cache: c.Expr[Cache[F, String, V]], config: c.Expr[MemoizationConfig], flags: c.Expr[Flags]): c.Tree = {
     commonMacroImpl(
-      cache,
+      config,
       { keyName =>
-        q"""$cache.cachingForMemoize($keyName)($ttl)($f)($flags)"""
+        q"""$cache.caching($keyName)($ttl)($f)($flags)"""
       }
     )
   }
 
   def memoizeFImpl[F[_], V: c.WeakTypeTag](
       ttl: c.Expr[Option[Duration]]
-  )(f: c.Tree)(cache: c.Expr[Cache[F, V]], flags: c.Expr[Flags]): c.Tree = {
+  )(f: c.Tree)(cache: c.Expr[Cache[F, String, V]], config: c.Expr[MemoizationConfig], flags: c.Expr[Flags]): c.Tree = {
     commonMacroImpl(
-      cache,
+      config,
       { keyName =>
-        q"""$cache.cachingForMemoizeF($keyName)($ttl)($f)($flags)"""
+        q"""$cache.cachingF($keyName)($ttl)($f)($flags)"""
       }
     )
   }
 
   private def commonMacroImpl[F[_], V: c.WeakTypeTag](
-      cache: c.Expr[Cache[F, V]],
+      config: c.Expr[MemoizationConfig],
       keyNameToCachingCall: (c.TermName) => c.Tree
   ): Tree = {
 
@@ -52,7 +53,7 @@ class Macros(val c: blackbox.Context) {
     val keyName     = createKeyName()
     val cachingCall = keyNameToCachingCall(keyName)
     val tree = q"""
-          val $keyName = $cache.config.memoization.toStringConverter.toString($classNameTree, $classParamssTree, $methodNameTree, $methodParamssTree)
+          val $keyName = $config.toStringConverter.toString($classNameTree, $classParamssTree, $methodNameTree, $methodParamssTree)
           $cachingCall
         """
     // println(showCode(tree))
diff --git a/modules/core/src/main/scala-2/scalacache/memoization/package.scala b/modules/core/src/main/scala-2/scalacache/memoization/package.scala
index 76ae3c5b..4b5918c8 100644
--- a/modules/core/src/main/scala-2/scalacache/memoization/package.scala
+++ b/modules/core/src/main/scala-2/scalacache/memoization/package.scala
@@ -34,7 +34,9 @@ package object memoization {
     * @return
     *   A result, either retrieved from the cache or calculated by executing the function `f`
     */
-  def memoize[F[_], V](ttl: Option[Duration])(f: => V)(implicit cache: Cache[F, V], flags: Flags): F[V] =
+  def memoize[F[_], V](ttl: Option[Duration])(
+      f: => V
+  )(implicit cache: Cache[F, String, V], config: MemoizationConfig, flags: Flags): F[V] =
     macro Macros.memoizeImpl[F, V]
 
   /** Perform the given operation and memoize its result to a cache before returning it. If the result is already in the
@@ -64,6 +66,6 @@ package object memoization {
     */
   def memoizeF[F[_], V](
       ttl: Option[Duration]
-  )(f: F[V])(implicit cache: Cache[F, V], flags: Flags): F[V] =
+  )(f: F[V])(implicit cache: Cache[F, String, V], config: MemoizationConfig, flags: Flags): F[V] =
     macro Macros.memoizeFImpl[F, V]
 }
diff --git a/modules/core/src/main/scala-3/scalacache/memoization/Macros.scala b/modules/core/src/main/scala-3/scalacache/memoization/Macros.scala
index e47dcea6..a6758d35 100644
--- a/modules/core/src/main/scala-3/scalacache/memoization/Macros.scala
+++ b/modules/core/src/main/scala-3/scalacache/memoization/Macros.scala
@@ -9,21 +9,23 @@ object Macros {
   def memoizeImpl[F[_], V](
       ttl: Expr[Option[Duration]],
       f: Expr[V],
-      cache: Expr[Cache[F, V]],
+      cache: Expr[Cache[F, String, V]],
+      config: Expr[MemoizationConfig],
       flags: Expr[Flags]
   )(using Quotes, Type[F], Type[V]): Expr[F[V]] =
-    commonMacroImpl(cache, keyName => '{ ${ cache }.cachingForMemoize(${ keyName })(${ ttl })(${ f })(${ flags }) })
+    commonMacroImpl(config, keyName => '{ ${ cache }.caching(${ keyName })(${ ttl })(${ f })(${ flags }) })
 
   def memoizeFImpl[F[_], V](
       ttl: Expr[Option[Duration]],
       f: Expr[F[V]],
-      cache: Expr[Cache[F, V]],
+      cache: Expr[Cache[F, String, V]],
+      config: Expr[MemoizationConfig],
       flags: Expr[Flags]
   )(using Quotes, Type[F], Type[V]): Expr[F[V]] =
-    commonMacroImpl(cache, keyName => '{ ${ cache }.cachingForMemoizeF(${ keyName })(${ ttl })(${ f })(${ flags }) })
+    commonMacroImpl(config, keyName => '{ ${ cache }.cachingF(${ keyName })(${ ttl })(${ f })(${ flags }) })
 
   private def commonMacroImpl[F[_], V](
-      cache: Expr[Cache[F, V]],
+      config: Expr[MemoizationConfig],
       keyNameToCachingCall: Expr[String] => Expr[F[V]]
   )(using Quotes, Type[F], Type[V]): Expr[F[V]] = {
     import quotes.reflect.*
@@ -69,7 +71,7 @@ object Macros {
     val classParamExpr: Expr[IndexedSeq[IndexedSeq[Any]]] = traverse(classParams map traverse)
 
     val keyValue: Expr[String] = '{
-      ${ cache }.config.memoization.toStringConverter.toString(
+      ${ config }.toStringConverter.toString(
         ${ Expr(classdefSymbol.fullName) },
         $classParamExpr,
         ${ Expr(defdef.name) },
diff --git a/modules/core/src/main/scala-3/scalacache/memoization/package.scala b/modules/core/src/main/scala-3/scalacache/memoization/package.scala
index 184b138a..0a2f9d4d 100644
--- a/modules/core/src/main/scala-3/scalacache/memoization/package.scala
+++ b/modules/core/src/main/scala-3/scalacache/memoization/package.scala
@@ -32,8 +32,10 @@ package object memoization {
     * @return
     *   A result, either retrieved from the cache or calculated by executing the function `f`
     */
-  inline def memoize[F[_], V](ttl: Option[Duration])(f: => V)(implicit cache: Cache[F, V], flags: Flags): F[V] =
-    ${ Macros.memoizeImpl[F, V]('ttl, 'f, 'cache, 'flags) }
+  inline def memoize[F[_], V](ttl: Option[Duration])(
+      f: => V
+  )(implicit cache: Cache[F, String, V], config: MemoizationConfig, flags: Flags): F[V] =
+    ${ Macros.memoizeImpl[F, V]('ttl, 'f, 'cache, 'config, 'flags) }
 
   /** Perform the given operation and memoize its result to a cache before returning it. If the result is already in the
     * cache, return it without performing the operation.
@@ -60,6 +62,8 @@ package object memoization {
     * @return
     *   A result, either retrieved from the cache or calculated by executing the function `f`
     */
-  inline def memoizeF[F[_], V](ttl: Option[Duration])(f: F[V])(implicit cache: Cache[F, V], flags: Flags): F[V] =
-    ${ Macros.memoizeFImpl[F, V]('ttl, 'f, 'cache, 'flags) }
+  inline def memoizeF[F[_], V](ttl: Option[Duration])(
+      f: F[V]
+  )(implicit cache: Cache[F, String, V], config: MemoizationConfig, flags: Flags): F[V] =
+    ${ Macros.memoizeFImpl[F, V]('ttl, 'f, 'cache, 'config, 'flags) }
 }
diff --git a/modules/core/src/main/scala/scalacache/AbstractCache.scala b/modules/core/src/main/scala/scalacache/AbstractCache.scala
index e33d7c59..6005519e 100644
--- a/modules/core/src/main/scala/scalacache/AbstractCache.scala
+++ b/modules/core/src/main/scala/scalacache/AbstractCache.scala
@@ -9,23 +9,25 @@ import cats.MonadError
 import cats.effect.Sync
 import cats.Applicative
 
-/** An abstract implementation of [[CacheAlg]] that takes care of some things that are common across all concrete
+/** An abstract implementation of [[Cache]] that takes care of some things that are common across all concrete
   * implementations.
   *
-  * If you are writing a cache implementation, you probably want to extend this trait rather than extending [[CacheAlg]]
+  * If you are writing a cache implementation, you probably want to extend this trait rather than extending [[Cache]]
   * directly.
   *
+  * @tparam K
+  *   The type of keys stored in the cache.
   * @tparam V
-  *   The value of types stored in the cache.
+  *   The type of values stored in the cache.
   */
-trait AbstractCache[F[_], V] extends Cache[F, V] with LoggingSupport[F] {
+trait AbstractCache[F[_], K, V] extends Cache[F, K, V] with LoggingSupport[F, K] {
 
   protected implicit def F: Sync[F]
   // GET
 
-  protected def doGet(key: String): F[Option[V]]
+  protected def doGet(key: K): F[Option[V]]
 
-  private def checkFlagsAndGet(key: String)(implicit flags: Flags): F[Option[V]] = {
+  private def checkFlagsAndGet(key: K)(implicit flags: Flags): F[Option[V]] = {
     if (flags.readsEnabled) {
       doGet(key)
     } else
@@ -36,16 +38,15 @@ trait AbstractCache[F[_], V] extends Cache[F, V] with LoggingSupport[F] {
         .as(None)
   }
 
-  final override def get(keyParts: Any*)(implicit flags: Flags): F[Option[V]] = {
-    val key = toKey(keyParts: _*)
+  final override def get(key: K)(implicit flags: Flags): F[Option[V]] = {
     checkFlagsAndGet(key)
   }
 
   // PUT
 
-  protected def doPut(key: String, value: V, ttl: Option[Duration]): F[Unit]
+  protected def doPut(key: K, value: V, ttl: Option[Duration]): F[Unit]
 
-  private def checkFlagsAndPut(key: String, value: V, ttl: Option[Duration])(implicit
+  private def checkFlagsAndPut(key: K, value: V, ttl: Option[Duration])(implicit
       flags: Flags
   ): F[Unit] = {
     if (flags.writesEnabled) {
@@ -57,19 +58,18 @@ trait AbstractCache[F[_], V] extends Cache[F, V] with LoggingSupport[F] {
   }
 
   final override def put(
-      keyParts: Any*
+      key: K
   )(value: V, ttl: Option[Duration])(implicit flags: Flags): F[Unit] = {
-    val key       = toKey(keyParts: _*)
     val finiteTtl = ttl.filter(_.isFinite) // discard Duration.Inf, Duration.Undefined
     checkFlagsAndPut(key, value, finiteTtl)
   }
 
   // REMOVE
 
-  protected def doRemove(key: String): F[Unit]
+  protected def doRemove(key: K): F[Unit]
 
-  final override def remove(keyParts: Any*): F[Unit] =
-    doRemove(toKey(keyParts: _*))
+  final override def remove(key: K): F[Unit] =
+    doRemove(key)
 
   // REMOVE ALL
 
@@ -81,42 +81,12 @@ trait AbstractCache[F[_], V] extends Cache[F, V] with LoggingSupport[F] {
   // CACHING
 
   final override def caching(
-      keyParts: Any*
-  )(ttl: Option[Duration] = None)(f: => V)(implicit flags: Flags): F[V] = {
-    val key = toKey(keyParts: _*)
-    _caching(key, ttl, f)
-  }
+      key: K
+  )(ttl: Option[Duration] = None)(f: => V)(implicit flags: Flags): F[V] = cachingF(key)(ttl)(Sync[F].delay(f))
 
   override def cachingF(
-      keyParts: Any*
+      key: K
   )(ttl: Option[Duration] = None)(f: F[V])(implicit flags: Flags): F[V] = {
-    val key = toKey(keyParts: _*)
-    _cachingF(key, ttl, f)
-  }
-
-  // MEMOIZE
-
-  override def cachingForMemoize(
-      baseKey: String
-  )(ttl: Option[Duration] = None)(f: => V)(implicit flags: Flags): F[V] = {
-    val key = config.cacheKeyBuilder.stringToCacheKey(baseKey)
-    _caching(key, ttl, f)
-  }
-
-  override def cachingForMemoizeF(
-      baseKey: String
-  )(ttl: Option[Duration])(f: F[V])(implicit flags: Flags): F[V] = {
-    val key = config.cacheKeyBuilder.stringToCacheKey(baseKey)
-    _cachingF(key, ttl, f)
-  }
-
-  private def _caching(key: String, ttl: Option[Duration], f: => V)(implicit
-      flags: Flags
-  ): F[V] = _cachingF(key, ttl, Sync[F].delay(f))
-
-  private def _cachingF(key: String, ttl: Option[Duration], f: => F[V])(implicit
-      flags: Flags
-  ): F[V] = {
     checkFlagsAndGet(key)
       .handleErrorWith { e =>
         logger
@@ -136,8 +106,4 @@ trait AbstractCache[F[_], V] extends Cache[F, V] with LoggingSupport[F] {
           }
       }
   }
-
-  private def toKey(keyParts: Any*): String =
-    config.cacheKeyBuilder.toCacheKey(keyParts)
-
 }
diff --git a/modules/core/src/main/scala/scalacache/Cache.scala b/modules/core/src/main/scala/scalacache/Cache.scala
index 735cf1e4..5abdc3bc 100644
--- a/modules/core/src/main/scala/scalacache/Cache.scala
+++ b/modules/core/src/main/scala/scalacache/Cache.scala
@@ -1,22 +1,93 @@
 package scalacache
 
 import scala.concurrent.duration.Duration
+
 import scala.language.higherKinds
 
-//todo merge with alg?
-trait Cache[F[_], V] extends CacheAlg[F, V] {
+/** Abstract algebra describing the operations a cache can perform
+  *
+  * @tparam F
+  *   The effect monad in which all cache operations will be performed.
+  * @tparam K
+  *   The type of keys stored in the cache.
+  * @tparam V
+  *   The type of values stored in the cache.
+  */
+trait Cache[F[_], K, V] {
+
+  /** Get a value from the cache
+    *
+    * @param key
+    *   The cache key
+    * @param flags
+    *   Flags used to conditionally alter the behaviour of ScalaCache
+    * @return
+    *   The appropriate value, if it was found in the cache
+    */
+  def get(key: K)(implicit flags: Flags): F[Option[V]]
+
+  /** Insert a value into the cache, optionally setting a TTL (time-to-live)
+    *
+    * @param key
+    *   The cache key
+    * @param value
+    *   The value to insert
+    * @param ttl
+    *   The time-to-live. The cache entry will expire after this time has elapsed.
+    * @param flags
+    *   Flags used to conditionally alter the behaviour of ScalaCache
+    */
+  def put(key: K)(value: V, ttl: Option[Duration] = None)(implicit flags: Flags): F[Unit]
+
+  /** Remove the given key and its associated value from the cache, if it exists. If the key is not in the cache, do
+    * nothing.
+    *
+    * @param key
+    *   The cache key
+    */
+  def remove(key: K): F[Unit]
 
-  def config: CacheConfig
+  /** Delete the entire contents of the cache. Use wisely!
+    */
+  def removeAll: F[Unit]
 
-  // Optimised methods for use by memoize: we know the key will be a single string so we can avoid some work.
-  // These are public because calls to them are generated by the memoize macro.
+  /** Get a value from the cache if it exists. Otherwise compute it, insert it into the cache, and return it.
+    *
+    * @param key
+    *   The cache key
+    * @param ttl
+    *   The time-to-live to use when inserting into the cache. The cache entry will expire after this time has elapsed.
+    * @param f
+    *   A block that computes the value
+    * @param flags
+    *   Flags used to conditionally alter the behaviour of ScalaCache
+    * @return
+    *   The value, either retrieved from the cache or computed
+    */
+  def caching(key: K)(ttl: Option[Duration])(f: => V)(implicit flags: Flags): F[V]
 
-  def cachingForMemoize(baseKey: String)(ttl: Option[Duration])(
-      f: => V
-  )(implicit flags: Flags): F[V]
+  /** Get a value from the cache if it exists. Otherwise compute it, insert it into the cache, and return it.
+    *
+    * @param key
+    *   The cache key
+    * @param ttl
+    *   The time-to-live to use when inserting into the cache. The cache entry will expire after this time has elapsed.
+    * @param f
+    *   A block that computes the value wrapped in a container
+    * @param flags
+    *   Flags used to conditionally alter the behaviour of ScalaCache
+    * @return
+    *   The value, either retrieved from the cache or computed
+    */
+  def cachingF(key: K)(ttl: Option[Duration])(f: F[V])(implicit flags: Flags): F[V]
 
-  def cachingForMemoizeF(baseKey: String)(ttl: Option[Duration])(
-      f: F[V]
-  )(implicit flags: Flags): F[V]
+  /** You should call this when you have finished using this Cache. (e.g. when your application shuts down)
+    *
+    * It will take care of gracefully shutting down the underlying cache client.
+    *
+    * Note that you should not try to use this Cache instance after you have called this method.
+    */
+  // TODO: Replace with Resource-based API?
+  def close: F[Unit]
 
 }
diff --git a/modules/core/src/main/scala/scalacache/CacheConfig.scala b/modules/core/src/main/scala/scalacache/CacheConfig.scala
deleted file mode 100644
index ba915585..00000000
--- a/modules/core/src/main/scala/scalacache/CacheConfig.scala
+++ /dev/null
@@ -1,14 +0,0 @@
-package scalacache
-
-import scalacache.memoization.MemoizationConfig
-
-case class CacheConfig(
-    cacheKeyBuilder: CacheKeyBuilder = DefaultCacheKeyBuilder(),
-    memoization: MemoizationConfig = MemoizationConfig()
-)
-
-object CacheConfig {
-
-  implicit val defaultCacheConfig: CacheConfig = CacheConfig()
-
-}
diff --git a/modules/core/src/main/scala/scalacache/CacheKeyBuilder.scala b/modules/core/src/main/scala/scalacache/CacheKeyBuilder.scala
deleted file mode 100644
index 392e4017..00000000
--- a/modules/core/src/main/scala/scalacache/CacheKeyBuilder.scala
+++ /dev/null
@@ -1,62 +0,0 @@
-package scalacache
-
-trait CacheKeyBuilder {
-
-  /** Build a String from the given parts to use as a cache key
-    */
-  def toCacheKey(parts: Seq[Any]): String
-
-  /** Build a cache key by prepending the configured prefix, if there is one
-    */
-  def stringToCacheKey(key: String): String
-
-}
-
-case class DefaultCacheKeyBuilder(keyPrefix: Option[String] = None, separator: String = ":") extends CacheKeyBuilder {
-
-  /** Build a String from the given parts, separating them using the configured separator string. Prepends the prefix if
-    * there is one.
-    */
-  override def toCacheKey(parts: Seq[Any]): String = {
-    // Implementation note: the type of `parts` will be `WrappedArray` when called from the package object,
-    // so random access is O(1).
-    val sb = new StringBuilder(128)
-
-    // Add the key prefix if there is one
-    keyPrefix match {
-      case Some(prefix) =>
-        sb.append(prefix)
-        sb.append(separator)
-      case None => // do nothing
-    }
-
-    var i = 0
-
-    // Add all key parts except the last one, with the separator after each one
-    while (i < parts.size - 1) {
-      sb.append(parts(i))
-      sb.append(separator)
-      i += 1
-    }
-    // Add the final key part
-    if (i < parts.size) {
-      sb.append(parts(i))
-    }
-
-    sb.toString
-  }
-
-  override def stringToCacheKey(key: String): String = {
-    keyPrefix match {
-      case Some(prefix) =>
-        val sb = new StringBuilder(prefix.length + separator.length + key.length)
-        sb.append(prefix)
-        sb.append(separator)
-        sb.append(key)
-        sb.toString
-      case None =>
-        key // just return the key as is
-    }
-  }
-
-}
diff --git a/modules/core/src/main/scala/scalacache/LoggingSupport.scala b/modules/core/src/main/scala/scalacache/LoggingSupport.scala
index db366a19..7f0ac279 100644
--- a/modules/core/src/main/scala/scalacache/LoggingSupport.scala
+++ b/modules/core/src/main/scala/scalacache/LoggingSupport.scala
@@ -10,7 +10,7 @@ import cats.implicits._
 
 /** Helper methods for logging
   */
-trait LoggingSupport[F[_]] {
+trait LoggingSupport[F[_], K] {
   protected def logger: Logger[F]
   protected implicit def F: Monad[F]
 
@@ -23,7 +23,7 @@ trait LoggingSupport[F[_]] {
     * @tparam A
     *   the type of the cache value
     */
-  protected def logCacheHitOrMiss[A](key: String, result: Option[A]): F[Unit] =
+  protected def logCacheHitOrMiss[A](key: K, result: Option[A]): F[Unit] =
     logger.ifDebugEnabled {
       val hitOrMiss = result.map(_ => "hit") getOrElse "miss"
       logger.debug(s"Cache $hitOrMiss for key $key")
@@ -36,7 +36,7 @@ trait LoggingSupport[F[_]] {
     * @param ttl
     *   the TTL of the inserted entry
     */
-  protected def logCachePut(key: String, ttl: Option[Duration]): F[Unit] =
+  protected def logCachePut(key: K, ttl: Option[Duration]): F[Unit] =
     logger.ifDebugEnabled {
       val ttlMsg = ttl.map(d => s" with TTL ${d.toMillis} ms") getOrElse ""
       logger.debug(s"Inserted value into cache with key $key$ttlMsg")
diff --git a/modules/core/src/main/scala/scalacache/memoization/MemoizationConfig.scala b/modules/core/src/main/scala/scalacache/memoization/MemoizationConfig.scala
index 0b79c201..f47febee 100644
--- a/modules/core/src/main/scala/scalacache/memoization/MemoizationConfig.scala
+++ b/modules/core/src/main/scala/scalacache/memoization/MemoizationConfig.scala
@@ -5,6 +5,10 @@ package scalacache.memoization
   * @param toStringConverter
   *   converter for generating a String cache key from information about a method call
   */
-case class MemoizationConfig(
-    toStringConverter: MethodCallToStringConverter = MethodCallToStringConverter.excludeClassConstructorParams
-)
+case class MemoizationConfig(toStringConverter: MethodCallToStringConverter)
+
+object MemoizationConfig {
+  implicit val defaultMemoizationConfig: MemoizationConfig = MemoizationConfig(
+    MethodCallToStringConverter.excludeClassConstructorParams
+  )
+}
diff --git a/modules/core/src/main/scala/scalacache/package.scala b/modules/core/src/main/scala/scalacache/package.scala
index df2c040b..c96cd2d9 100644
--- a/modules/core/src/main/scala/scalacache/package.scala
+++ b/modules/core/src/main/scala/scalacache/package.scala
@@ -5,9 +5,6 @@ package object scalacache {
 
   /** Get the value corresponding to the given key from the cache.
     *
-    * @param keyParts
-    *   Data to be used to generate the cache key. This could be as simple as just a single String. See
-    *   [[CacheKeyBuilder]].
     * @param cache
     *   The cache
     * @param mode
@@ -21,8 +18,8 @@ package object scalacache {
     * @return
     *   the value, if there is one
     */
-  def get[F[_], V](keyParts: Any*)(implicit cache: Cache[F, V], flags: Flags): F[Option[V]] =
-    cache.get(keyParts: _*)
+  def get[F[_], K, V](key: K)(implicit cache: Cache[F, K, V], flags: Flags): F[Option[V]] =
+    cache.get(key)
 
   /** Insert the given key-value pair into the cache, with an optional Time To Live.
     *
@@ -46,10 +43,10 @@ package object scalacache {
     * @tparam V
     *   The type of the corresponding value
     */
-  def put[F[_], V](
-      keyParts: Any*
-  )(value: V, ttl: Option[Duration] = None)(implicit cache: Cache[F, V], flags: Flags): F[Unit] =
-    cache.put(keyParts: _*)(value, ttl)
+  def put[F[_], K, V](
+      key: K
+  )(value: V, ttl: Option[Duration] = None)(implicit cache: Cache[F, K, V], flags: Flags): F[Unit] =
+    cache.put(key)(value, ttl)
 
   /** Remove the given key and its associated value from the cache, if it exists. If the key is not in the cache, do
     * nothing.
@@ -68,11 +65,11 @@ package object scalacache {
     * @tparam V
     *   The type of the value to be removed
     */
-  def remove[F[_], V](keyParts: Any*)(implicit cache: Cache[F, V]): F[Unit] =
-    cache.remove(keyParts: _*)
+  def remove[F[_], K, V](key: K)(implicit cache: Cache[F, K, V]): F[Unit] =
+    cache.remove(key)
 
-  final class RemoveAll[V] {
-    def apply[F[_]]()(implicit cache: Cache[F, V]): F[Unit] = cache.removeAll
+  final class RemoveAll[K, V] {
+    def apply[F[_]]()(implicit cache: Cache[F, K, V]): F[Unit] = cache.removeAll
   }
 
   /** Remove all values from the cache.
@@ -80,7 +77,7 @@ package object scalacache {
     * @tparam V
     *   The type of values to be removed
     */
-  def removeAll[V]: RemoveAll[V] = new RemoveAll[V]
+  def removeAll[K, V]: RemoveAll[K, V] = new RemoveAll[K, V]
 
   /** Wrap the given block with a caching decorator. First look in the cache. If the value is found, then return it
     * immediately. Otherwise run the block and save the result in the cache before returning it.
@@ -108,10 +105,10 @@ package object scalacache {
     * @return
     *   The result, either retrived from the cache or returned by the block
     */
-  def caching[F[_], V](
-      keyParts: Any*
-  )(ttl: Option[Duration])(f: => V)(implicit cache: Cache[F, V], flags: Flags): F[V] =
-    cache.caching(keyParts: _*)(ttl)(f)
+  def caching[F[_], K, V](
+      key: K
+  )(ttl: Option[Duration])(f: => V)(implicit cache: Cache[F, K, V], flags: Flags): F[V] =
+    cache.caching(key)(ttl)(f)
 
   /** Wrap the given block with a caching decorator. First look in the cache. If the value is found, then return it
     * immediately. Otherwise run the block and save the result in the cache before returning it.
@@ -139,8 +136,8 @@ package object scalacache {
     * @return
     *   The result, either retrived from the cache or returned by the block
     */
-  def cachingF[F[_], V](
-      keyParts: Any*
-  )(ttl: Option[Duration])(f: => F[V])(implicit cache: Cache[F, V], flags: Flags): F[V] =
-    cache.cachingF(keyParts: _*)(ttl)(f)
+  def cachingF[F[_], K, V](
+      key: K
+  )(ttl: Option[Duration])(f: => F[V])(implicit cache: Cache[F, K, V], flags: Flags): F[V] =
+    cache.cachingF(key)(ttl)(f)
 }
diff --git a/modules/core/src/main/scala/scalacache/serialization/Codec.scala b/modules/core/src/main/scala/scalacache/serialization/Codec.scala
index 5a603ea5..84b77581 100644
--- a/modules/core/src/main/scala/scalacache/serialization/Codec.scala
+++ b/modules/core/src/main/scala/scalacache/serialization/Codec.scala
@@ -4,15 +4,23 @@ import scala.annotation.implicitNotFound
 import scala.language.implicitConversions
 import scala.util.{Failure, Success, Try}
 
+trait Encoder[L, R] {
+  def encode(left: L): R
+}
+
+trait Decoder[L, R] {
+  def decode(right: R): Codec.DecodingResult[L]
+}
+
 /** Represents a type class that needs to be implemented for serialization/deserialization to work.
   */
-@implicitNotFound(msg = """Could not find any Codecs for type ${A}.
+@implicitNotFound(msg = """Could not find any Codecs for types ${L, R}.
 If you would like to serialize values in a binary format, please import the binary codec:
 
 import scalacache.serialization.binary._
 
 If you would like to serialize values as JSON using circe, please import the circe codec
-and provide a circe Encoder[${A}] and Decoder[${A}], e.g.:
+and provide a circe Encoder[${L}] and Decoder[${L}], e.g.:
 
 import scalacache.serialization.circe._
 import io.circe.generic.auto._
@@ -20,18 +28,18 @@ import io.circe.generic.auto._
 You will need a dependency on the scalacache-circe module.
 
 See the documentation for more details on codecs.""")
-trait Codec[A] {
-  def encode(value: A): Array[Byte]
-  def decode(bytes: Array[Byte]): Codec.DecodingResult[A]
+trait Codec[L, R] extends Encoder[L, R] with Decoder[L, R] {
+  override def encode(left: L): R
+  override def decode(right: R): Codec.DecodingResult[L]
 }
 
 /** For simple primitives, we provide lightweight Codecs for ease of use.
   */
 object Codec {
 
-  type DecodingResult[A] = Either[FailedToDecode, A]
+  type DecodingResult[T] = Either[FailedToDecode, T]
 
-  def tryDecode[A](f: => A): DecodingResult[A] =
+  def tryDecode[T](f: => T): DecodingResult[T] =
     Try(f) match {
       case Success(a) => Right(a)
       case Failure(e) => Left(FailedToDecode(e))
diff --git a/modules/core/src/main/scala/scalacache/serialization/binary/BinaryAnyRefCodecs.scala b/modules/core/src/main/scala/scalacache/serialization/binary/BinaryAnyRefCodecs.scala
index 403b2e4a..d3e62108 100644
--- a/modules/core/src/main/scala/scalacache/serialization/binary/BinaryAnyRefCodecs.scala
+++ b/modules/core/src/main/scala/scalacache/serialization/binary/BinaryAnyRefCodecs.scala
@@ -9,7 +9,7 @@ trait BinaryAnyRefCodecs_1 {
   String and Array[Byte] extend java.io.Serializable,
   so this implicit needs to be lower priority than those in BinaryPrimitiveCodecs
    */
-  implicit def anyRefBinaryCodec[S <: java.io.Serializable](implicit ev: ClassTag[S]): Codec[S] =
+  implicit def anyRefBinaryCodec[S <: java.io.Serializable](implicit ev: ClassTag[S]): BinaryCodec[S] =
     new JavaSerializationAnyRefCodec[S](ev)
 
 }
diff --git a/modules/core/src/main/scala/scalacache/serialization/binary/BinaryCodec.scala b/modules/core/src/main/scala/scalacache/serialization/binary/BinaryCodec.scala
new file mode 100644
index 00000000..91a96006
--- /dev/null
+++ b/modules/core/src/main/scala/scalacache/serialization/binary/BinaryCodec.scala
@@ -0,0 +1,16 @@
+package scalacache.serialization.binary
+
+import scalacache.serialization.{Codec, Decoder, Encoder}
+
+trait BinaryEncoder[T] extends Encoder[T, Array[Byte]] {
+  override def encode(value: T): Array[Byte]
+}
+
+trait BinaryDecoder[T] extends Decoder[T, Array[Byte]] {
+  override def decode(value: Array[Byte]): Codec.DecodingResult[T]
+}
+
+trait BinaryCodec[T] extends Codec[T, Array[Byte]] with BinaryEncoder[T] with BinaryDecoder[T] {
+  override def encode(value: T): Array[Byte]
+  override def decode(bytes: Array[Byte]): Codec.DecodingResult[T]
+}
diff --git a/modules/core/src/main/scala/scalacache/serialization/binary/BinaryPrimitiveCodecs.scala b/modules/core/src/main/scala/scalacache/serialization/binary/BinaryPrimitiveCodecs.scala
index 401d74b2..67342f4f 100644
--- a/modules/core/src/main/scala/scalacache/serialization/binary/BinaryPrimitiveCodecs.scala
+++ b/modules/core/src/main/scala/scalacache/serialization/binary/BinaryPrimitiveCodecs.scala
@@ -1,6 +1,5 @@
 package scalacache.serialization.binary
 
-import scalacache.serialization.Codec
 import scalacache.serialization.Codec._
 
 /** Codecs for all the Java primitive types, plus String and Array[Byte]
@@ -9,7 +8,7 @@ import scalacache.serialization.Codec._
   */
 trait BinaryPrimitiveCodecs {
 
-  implicit object IntBinaryCodec extends Codec[Int] {
+  implicit object IntBinaryCodec extends BinaryCodec[Int] {
     def encode(value: Int): Array[Byte] =
       Array(
         (value >>> 24).asInstanceOf[Byte],
@@ -26,7 +25,7 @@ trait BinaryPrimitiveCodecs {
     )
   }
 
-  implicit object DoubleBinaryCodec extends Codec[Double] {
+  implicit object DoubleBinaryCodec extends BinaryCodec[Double] {
     import java.lang.{Double => JvmDouble}
     def encode(value: Double): Array[Byte] = {
       val l = JvmDouble.doubleToLongBits(value)
@@ -41,7 +40,7 @@ trait BinaryPrimitiveCodecs {
     }
   }
 
-  implicit object FloatBinaryCodec extends Codec[Float] {
+  implicit object FloatBinaryCodec extends BinaryCodec[Float] {
     import java.lang.{Float => JvmFloat}
     def encode(value: Float): Array[Byte] = {
       val i = JvmFloat.floatToIntBits(value)
@@ -56,7 +55,7 @@ trait BinaryPrimitiveCodecs {
     }
   }
 
-  implicit object LongBinaryCodec extends Codec[Long] {
+  implicit object LongBinaryCodec extends BinaryCodec[Long] {
     def encode(value: Long): Array[Byte] =
       Array(
         (value >>> 56).asInstanceOf[Byte],
@@ -81,7 +80,7 @@ trait BinaryPrimitiveCodecs {
     )
   }
 
-  implicit object BooleanBinaryCodec extends Codec[Boolean] {
+  implicit object BooleanBinaryCodec extends BinaryCodec[Boolean] {
     def encode(value: Boolean): Array[Byte] =
       Array((if (value) 1 else 0).asInstanceOf[Byte])
 
@@ -89,7 +88,7 @@ trait BinaryPrimitiveCodecs {
       tryDecode(data.isDefinedAt(0) && data(0) == 1)
   }
 
-  implicit object CharBinaryCodec extends Codec[Char] {
+  implicit object CharBinaryCodec extends BinaryCodec[Char] {
     def encode(value: Char): Array[Byte] = Array(
       (value >>> 8).asInstanceOf[Byte],
       value.asInstanceOf[Byte]
@@ -102,7 +101,7 @@ trait BinaryPrimitiveCodecs {
     )
   }
 
-  implicit object ShortBinaryCodec extends Codec[Short] {
+  implicit object ShortBinaryCodec extends BinaryCodec[Short] {
     def encode(value: Short): Array[Byte] = Array(
       (value >>> 8).asInstanceOf[Byte],
       value.asInstanceOf[Byte]
@@ -115,12 +114,12 @@ trait BinaryPrimitiveCodecs {
     )
   }
 
-  implicit object StringBinaryCodec extends Codec[String] {
+  implicit object StringBinaryCodec extends BinaryCodec[String] {
     def encode(value: String): Array[Byte]                = value.getBytes("UTF-8")
     def decode(data: Array[Byte]): DecodingResult[String] = tryDecode(new String(data, "UTF-8"))
   }
 
-  implicit object ArrayByteBinaryCodec extends Codec[Array[Byte]] {
+  implicit object ArrayByteBinaryCodec extends BinaryCodec[Array[Byte]] {
     def encode(value: Array[Byte]): Array[Byte]                = value
     def decode(data: Array[Byte]): DecodingResult[Array[Byte]] = Right(data)
   }
diff --git a/modules/core/src/main/scala/scalacache/serialization/binary/JavaSerializationAnyRefCodec.scala b/modules/core/src/main/scala/scalacache/serialization/binary/JavaSerializationAnyRefCodec.scala
index a2cedf5d..b6617f0e 100644
--- a/modules/core/src/main/scala/scalacache/serialization/binary/JavaSerializationAnyRefCodec.scala
+++ b/modules/core/src/main/scala/scalacache/serialization/binary/JavaSerializationAnyRefCodec.scala
@@ -11,7 +11,7 @@ import scalacache.serialization.{Codec, GenericCodecObjectInputStream}
   *
   * Credit: Shade @ https://github.com/alexandru/shade/blob/master/src/main/scala/shade/memcached/Codec.scala
   */
-class JavaSerializationAnyRefCodec[S <: Serializable](classTag: ClassTag[S]) extends Codec[S] {
+class JavaSerializationAnyRefCodec[S <: Serializable](classTag: ClassTag[S]) extends BinaryCodec[S] {
 
   def using[T <: Closeable, R](obj: T)(f: T => R): R =
     try f(obj)
diff --git a/modules/core/src/main/scala/scalacache/serialization/gzip/GzippingBinaryCodec.scala b/modules/core/src/main/scala/scalacache/serialization/gzip/GzippingBinaryCodec.scala
index d83aaf62..6f13cb41 100644
--- a/modules/core/src/main/scala/scalacache/serialization/gzip/GzippingBinaryCodec.scala
+++ b/modules/core/src/main/scala/scalacache/serialization/gzip/GzippingBinaryCodec.scala
@@ -2,8 +2,8 @@ package scalacache.serialization.gzip
 
 import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
 import java.util.zip.{GZIPInputStream, GZIPOutputStream}
-
 import scalacache.serialization.Codec.DecodingResult
+import scalacache.serialization.binary.BinaryCodec
 import scalacache.serialization.{Codec, FailedToDecode}
 
 object CompressingCodec {
@@ -24,7 +24,7 @@ object CompressingCodec {
 /** Mixing this into any Codec will automatically GZip the resulting Byte Array when serialising and handle un-Gzipping
   * when deserialising
   */
-trait GZippingBinaryCodec[A] extends Codec[A] {
+trait GZippingBinaryCodec[A] extends BinaryCodec[A] {
 
   import CompressingCodec._
 
diff --git a/modules/core/src/test/scala/issue42/Issue42Spec.scala b/modules/core/src/test/scala/issue42/Issue42Spec.scala
index 09ffbbce..dc52ebaf 100644
--- a/modules/core/src/test/scala/issue42/Issue42Spec.scala
+++ b/modules/core/src/test/scala/issue42/Issue42Spec.scala
@@ -15,7 +15,7 @@ class Issue42Spec extends AnyFlatSpec with Matchers {
   import concurrent.duration._
   import scala.language.postfixOps
 
-  implicit val cache: Cache[SyncIO, User] = new MockCache()
+  implicit val cache: Cache[SyncIO, String, User] = new MockCache()
 
   def generateNewName() = Random.alphanumeric.take(10).mkString
 
diff --git a/modules/core/src/test/scala/sample/Sample.scala b/modules/core/src/test/scala/sample/Sample.scala
index 95e274fe..f7f6aadd 100644
--- a/modules/core/src/test/scala/sample/Sample.scala
+++ b/modules/core/src/test/scala/sample/Sample.scala
@@ -15,7 +15,7 @@ case class User(id: Int, name: String)
 object Sample extends App {
 
   class UserRepository {
-    implicit val cache: Cache[IO, User] = new MockCache()
+    implicit val cache: Cache[IO, String, User] = new MockCache()
 
     def getUser(id: Int): IO[User] = memoizeF(None) {
       // Do DB lookup here...
diff --git a/modules/core/src/test/scala/scalacache/AbstractCacheSpec.scala b/modules/core/src/test/scala/scalacache/AbstractCacheSpec.scala
index 6bbdd541..2308407a 100644
--- a/modules/core/src/test/scala/scalacache/AbstractCacheSpec.scala
+++ b/modules/core/src/test/scala/scalacache/AbstractCacheSpec.scala
@@ -27,11 +27,6 @@ class AbstractCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter {
     cache.getCalledWithArgs(0) should be("foo")
   }
 
-  it should "use the CacheKeyBuilder to build the cache key" in {
-    cache.get("foo", 123).unsafeRunSync()
-    cache.getCalledWithArgs(0) should be("foo:123")
-  }
-
   it should "not call doGet on the concrete cache if cache reads are disabled" in {
     implicit val flags: Flags = Flags(readsEnabled = false)
     cache.get("foo").unsafeRunSync()
@@ -74,11 +69,6 @@ class AbstractCacheSpec extends AnyFlatSpec with Matchers with BeforeAndAfter {
     cache.removeCalledWithArgs(0) should be("baz")
   }
 
-  it should "concatenate key parts correctly" in {
-    cache.remove("hey", "yeah").unsafeRunSync()
-    cache.removeCalledWithArgs(0) should be("hey:yeah")
-  }
-
   behavior of "#caching"
 
   it should "run the block and cache its result with no TTL if the value is not found in the cache" in {
diff --git a/modules/core/src/test/scala/scalacache/DefaultCacheKeyBuilderSpec.scala b/modules/core/src/test/scala/scalacache/DefaultCacheKeyBuilderSpec.scala
deleted file mode 100644
index 79b6b604..00000000
--- a/modules/core/src/test/scala/scalacache/DefaultCacheKeyBuilderSpec.scala
+++ /dev/null
@@ -1,26 +0,0 @@
-package scalacache
-
-import org.scalatest.flatspec.AnyFlatSpec
-import org.scalatest.matchers.should.Matchers
-
-class DefaultCacheKeyBuilderSpec extends AnyFlatSpec with Matchers {
-
-  behavior of "Default cache key builder"
-
-  it should "Use a single string as-is" in {
-    DefaultCacheKeyBuilder().toCacheKey(Seq("abc")) should be("abc")
-  }
-
-  it should "Separate the parts using the configured separator" in {
-    DefaultCacheKeyBuilder(separator = "_").toCacheKey(Seq("abc", 123)) should be("abc_123")
-  }
-
-  it should "Prepend the key prefix if one is configured" in {
-    DefaultCacheKeyBuilder(keyPrefix = Some("foo")).toCacheKey(Seq("abc", 123)) should be("foo:abc:123")
-  }
-
-  it should "Prepend the key prefix to a single string if one is configured" in {
-    DefaultCacheKeyBuilder(keyPrefix = Some("foo")).stringToCacheKey("abc") should be("foo:abc")
-  }
-
-}
diff --git a/modules/core/src/test/scala/scalacache/memoization/CacheKeyExcludingConstructorParamsSpec.scala b/modules/core/src/test/scala/scalacache/memoization/CacheKeyExcludingConstructorParamsSpec.scala
index a74a1474..5e320911 100644
--- a/modules/core/src/test/scala/scalacache/memoization/CacheKeyExcludingConstructorParamsSpec.scala
+++ b/modules/core/src/test/scala/scalacache/memoization/CacheKeyExcludingConstructorParamsSpec.scala
@@ -1,25 +1,24 @@
 package scalacache.memoization
 
-import org.scalatest._
-
-import scalacache._
-import scalacache.memoization.MethodCallToStringConverter.excludeClassConstructorParams
 import cats.effect.SyncIO
 import org.scalatest.flatspec.AnyFlatSpec
+import scalacache.memoization.MethodCallToStringConverter.excludeClassConstructorParams
 
 class CacheKeyExcludingConstructorParamsSpec extends AnyFlatSpec with CacheKeySpecCommon { self =>
 
   behavior of "cache key generation for method memoization (not including constructor params in cache key)"
 
-  implicit val config: CacheConfig =
-    CacheConfig(memoization = MemoizationConfig(toStringConverter = excludeClassConstructorParams))
+  override implicit lazy val config: MemoizationConfig =
+    MemoizationConfig(toStringConverter = excludeClassConstructorParams)
 
   it should "not include the enclosing class's constructor params in the cache key" in {
     val instance1 = new ClassWithConstructorParams[SyncIO](50)
     instance1.cache = cache
+    instance1.config = config
 
     val instance2 = new ClassWithConstructorParams[SyncIO](100)
     instance2.cache = cache
+    instance2.config = config
 
     checkCacheKey("scalacache.memoization.ClassWithConstructorParams.foo(42)") {
       instance1.foo(42).unsafeRunSync()
@@ -62,7 +61,7 @@ class CacheKeyExcludingConstructorParamsSpec extends AnyFlatSpec with CacheKeySp
 
   it should "work for a method inside a trait" in {
     checkCacheKey("scalacache.memoization.ATrait.insideTrait(1)") {
-      new ATrait[SyncIO] { val cache = self.cache }.insideTrait(1).unsafeRunSync()
+      new ATrait[SyncIO] { val cache = self.cache; val config = self.config }.insideTrait(1).unsafeRunSync()
     }
   }
 
diff --git a/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingConstructorParamsSpec.scala b/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingConstructorParamsSpec.scala
index c80ed62e..dd0e7c1d 100644
--- a/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingConstructorParamsSpec.scala
+++ b/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingConstructorParamsSpec.scala
@@ -1,21 +1,20 @@
 package scalacache.memoization
 
-import org.scalatest._
-import scalacache._
-import scalacache.memoization.MethodCallToStringConverter._
 import cats.effect.SyncIO
 import org.scalatest.flatspec.AnyFlatSpec
+import scalacache.memoization.MethodCallToStringConverter._
 
 class CacheKeyIncludingConstructorParamsSpec extends AnyFlatSpec with CacheKeySpecCommon { self =>
 
   behavior of "cache key generation for method memoization (when including constructor params in cache key)"
 
-  implicit val config: CacheConfig =
-    CacheConfig(memoization = MemoizationConfig(toStringConverter = includeClassConstructorParams))
+  implicit override lazy val config: MemoizationConfig =
+    MemoizationConfig(toStringConverter = includeClassConstructorParams)
 
   it should "include the enclosing class's constructor params in the cache key" in {
     val instance = new ClassWithConstructorParams[SyncIO](50)
     instance.cache = cache
+    instance.config = config
 
     checkCacheKey("scalacache.memoization.ClassWithConstructorParams(50).foo(42)") {
       instance.foo(42).unsafeRunSync()
@@ -25,6 +24,7 @@ class CacheKeyIncludingConstructorParamsSpec extends AnyFlatSpec with CacheKeySp
   it should "exclude values of constructor params annotated with @cacheKeyExclude" in {
     val instance = new ClassWithExcludedConstructorParam[SyncIO](50, 10)
     instance.cache = cache
+    instance.config = config
 
     checkCacheKey("scalacache.memoization.ClassWithExcludedConstructorParam(50).foo(42)") {
       instance.foo(42).unsafeRunSync()
@@ -57,19 +57,20 @@ class CacheKeyIncludingConstructorParamsSpec extends AnyFlatSpec with CacheKeySp
 
   it should "work for a method inside a class" in {
     // The class's implicit param (the Cache) should be included in the cache key)
-    checkCacheKey(s"scalacache.memoization.AClass()(${cache.toString}).insideClass(1)") {
+    checkCacheKey(s"scalacache.memoization.AClass()(${cache.toString}, ${config.toString}).insideClass(1)") {
       new AClass[SyncIO]().insideClass(1).unsafeRunSync()
     }
   }
 
   it should "work for a method inside a trait" in {
     checkCacheKey("scalacache.memoization.ATrait.insideTrait(1)") {
-      new ATrait[SyncIO] { val cache = self.cache }.insideTrait(1).unsafeRunSync()
+      new ATrait[SyncIO] { val cache = self.cache; val config = self.config }.insideTrait(1).unsafeRunSync()
     }
   }
 
   it should "work for a method inside an object" in {
     AnObject.cache = this.cache
+    AnObject.config = this.config
     checkCacheKey("scalacache.memoization.AnObject.insideObject(1)") {
       AnObject.insideObject(1).unsafeRunSync()
     }
diff --git a/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingOnlyMethodParamsSpec.scala b/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingOnlyMethodParamsSpec.scala
index 322c1263..23f02825 100644
--- a/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingOnlyMethodParamsSpec.scala
+++ b/modules/core/src/test/scala/scalacache/memoization/CacheKeyIncludingOnlyMethodParamsSpec.scala
@@ -1,17 +1,13 @@
 package scalacache.memoization
 
-import org.scalatest._
-
-import scalacache._
-import scalacache.memoization.MethodCallToStringConverter.onlyMethodParams
 import org.scalatest.flatspec.AnyFlatSpec
+import scalacache.memoization.MethodCallToStringConverter.onlyMethodParams
 
 class CacheKeyIncludingOnlyMethodParamsSpec extends AnyFlatSpec with CacheKeySpecCommon {
 
   behavior of "cache key generation for method memoization (only including method params in cache key)"
 
-  implicit val config: CacheConfig =
-    CacheConfig(memoization = MemoizationConfig(toStringConverter = onlyMethodParams))
+  override implicit lazy val config: MemoizationConfig = MemoizationConfig(toStringConverter = onlyMethodParams)
 
   it should "include values of all arguments for all argument lists" in {
     checkCacheKey("(1, 2)(3, 4)") {
diff --git a/modules/core/src/test/scala/scalacache/memoization/CacheKeySpecCommon.scala b/modules/core/src/test/scala/scalacache/memoization/CacheKeySpecCommon.scala
index 9e7d5c47..9d6a2d8b 100644
--- a/modules/core/src/test/scala/scalacache/memoization/CacheKeySpecCommon.scala
+++ b/modules/core/src/test/scala/scalacache/memoization/CacheKeySpecCommon.scala
@@ -8,7 +8,7 @@ import org.scalatest.matchers.should.Matchers
 
 trait CacheKeySpecCommon extends Suite with Matchers with BeforeAndAfter {
 
-  implicit def config: CacheConfig
+  implicit lazy val config: MemoizationConfig = scalacache.memoization.MemoizationConfig.defaultMemoizationConfig
 
   implicit lazy val cache: MockCache[SyncIO, Int] = new MockCache()
 
@@ -50,7 +50,7 @@ trait CacheKeySpecCommon extends Suite with Matchers with BeforeAndAfter {
 
 }
 
-class AClass[F[_]]()(implicit cache: Cache[F, Int]) {
+class AClass[F[_]]()(implicit cache: Cache[F, String, Int], config: MemoizationConfig) {
   def insideClass(a: Int): F[Int] = memoize(None) {
     123
   }
@@ -70,7 +70,8 @@ class AClass[F[_]]()(implicit cache: Cache[F, Int]) {
 }
 
 trait ATrait[F[_]] {
-  implicit val cache: Cache[F, Int]
+  implicit val cache: Cache[F, String, Int]
+  implicit val config: MemoizationConfig
 
   def insideTrait(a: Int): F[Int] = memoize(None) {
     123
@@ -78,21 +79,24 @@ trait ATrait[F[_]] {
 }
 
 object AnObject {
-  implicit var cache: Cache[SyncIO, Int] = null
+  implicit var cache: Cache[SyncIO, String, Int] = null
+  implicit var config: MemoizationConfig         = null
   def insideObject(a: Int): SyncIO[Int] = memoize(None) {
     123
   }
 }
 
 class ClassWithConstructorParams[F[_]](b: Int) {
-  implicit var cache: Cache[F, Int] = null
+  implicit var cache: Cache[F, String, Int] = null
+  implicit var config: MemoizationConfig    = null
   def foo(a: Int): F[Int] = memoize(None) {
     a + b
   }
 }
 
 class ClassWithExcludedConstructorParam[F[_]](b: Int, @cacheKeyExclude c: Int) {
-  implicit var cache: Cache[F, Int] = null
+  implicit var cache: Cache[F, String, Int] = null
+  implicit var config: MemoizationConfig    = null
   def foo(a: Int): F[Int] = memoize(None) {
     a + b + c
   }
diff --git a/modules/core/src/test/scala/scalacache/memoization/MemoizeSpec.scala b/modules/core/src/test/scala/scalacache/memoization/MemoizeSpec.scala
index 426e86f5..a5cc4c4c 100644
--- a/modules/core/src/test/scala/scalacache/memoization/MemoizeSpec.scala
+++ b/modules/core/src/test/scala/scalacache/memoization/MemoizeSpec.scala
@@ -257,7 +257,10 @@ class MemoizeSpec extends AnyFlatSpec with Matchers {
     }
   }
 
-  class MyMockClass[F[_]](dbCall: Int => String)(implicit val cache: Cache[F, String], flags: Flags) {
+  class MyMockClass[F[_]](dbCall: Int => String)(implicit
+      val cache: Cache[F, String, String],
+      flags: Flags
+  ) {
 
     def myLongRunningMethod(a: Int, b: String): F[String] = memoize(None) {
       dbCall(a)
@@ -270,7 +273,7 @@ class MemoizeSpec extends AnyFlatSpec with Matchers {
   }
 
   class MyMockClassWithTry[F[_]](dbCall: Int => String)(implicit
-      cache: Cache[F, String],
+      cache: Cache[F, String, String],
       F: Sync[F],
       flags: Flags
   ) {
diff --git a/modules/core/src/test/scala/scalacache/memoization/pkg/package.scala b/modules/core/src/test/scala/scalacache/memoization/pkg/package.scala
index d0557ff1..c107dc85 100644
--- a/modules/core/src/test/scala/scalacache/memoization/pkg/package.scala
+++ b/modules/core/src/test/scala/scalacache/memoization/pkg/package.scala
@@ -6,7 +6,7 @@ package object pkg {
 
   import cats.effect.SyncIO
 
-  implicit var cache: Cache[SyncIO, Int] = null
+  implicit var cache: Cache[SyncIO, String, Int] = null
 
   def insidePackageObject(a: Int): SyncIO[Int] = memoize(None) {
     123
diff --git a/modules/core/src/test/scala/scalacache/mocks.scala b/modules/core/src/test/scala/scalacache/mocks.scala
index dfd8e954..8443cd0d 100644
--- a/modules/core/src/test/scala/scalacache/mocks.scala
+++ b/modules/core/src/test/scala/scalacache/mocks.scala
@@ -1,15 +1,14 @@
 package scalacache
 
+import cats.effect.Sync
 import scalacache.logging.Logger
+import scalacache.memoization.MemoizationConfig
+
 import scala.collection.mutable.ArrayBuffer
 import scala.concurrent.duration.Duration
 import scala.language.higherKinds
-import cats.Applicative
-import cats.effect.Sync
-import cats.MonadError
-import cats.Defer
 
-class EmptyCache[F[_], V](implicit val F: Sync[F], val config: CacheConfig) extends AbstractCache[F, V] {
+class EmptyCache[F[_], V](implicit val F: Sync[F], val config: MemoizationConfig) extends AbstractCache[F, String, V] {
 
   override protected def logger = Logger.getLogger("EmptyCache")
 
@@ -29,7 +28,8 @@ class EmptyCache[F[_], V](implicit val F: Sync[F], val config: CacheConfig) exte
 
 }
 
-class FullCache[F[_], V](value: V)(implicit val F: Sync[F], val config: CacheConfig) extends AbstractCache[F, V] {
+class FullCache[F[_], V](value: V)(implicit val F: Sync[F], val config: MemoizationConfig)
+    extends AbstractCache[F, String, V] {
 
   override protected def logger = Logger.getLogger("FullCache")
 
@@ -49,7 +49,8 @@ class FullCache[F[_], V](value: V)(implicit val F: Sync[F], val config: CacheCon
 
 }
 
-class ErrorRaisingCache[F[_], V](implicit val F: Sync[F], val config: CacheConfig) extends AbstractCache[F, V] {
+class ErrorRaisingCache[F[_], V](implicit val F: Sync[F], val config: MemoizationConfig)
+    extends AbstractCache[F, String, V] {
 
   override protected val logger = Logger.getLogger("FullCache")
 
@@ -71,7 +72,7 @@ class ErrorRaisingCache[F[_], V](implicit val F: Sync[F], val config: CacheConfi
 
 /** A mock cache for use in tests and samples. Does not support TTL.
   */
-class MockCache[F[_], V](implicit val F: Sync[F], val config: CacheConfig) extends AbstractCache[F, V] {
+class MockCache[F[_], V](implicit val F: Sync[F], val config: MemoizationConfig) extends AbstractCache[F, String, V] {
 
   override protected def logger = Logger.getLogger("MockCache")
 
@@ -96,7 +97,7 @@ class MockCache[F[_], V](implicit val F: Sync[F], val config: CacheConfig) exten
 /** A cache that keeps track of the arguments it was called with. Useful for tests. Designed to be mixed in as a
   * stackable trait.
   */
-trait LoggingCache[F[_], V] extends AbstractCache[F, V] {
+trait LoggingCache[F[_], V] extends AbstractCache[F, String, V] {
   val F: Sync[F]
 
   var (getCalledWithArgs, putCalledWithArgs, removeCalledWithArgs) =
diff --git a/modules/docs/src/main/mdoc/docs/cache-implementations.md b/modules/docs/src/main/mdoc/docs/cache-implementations.md
index 515a9500..8852edf6 100644
--- a/modules/docs/src/main/mdoc/docs/cache-implementations.md
+++ b/modules/docs/src/main/mdoc/docs/cache-implementations.md
@@ -21,7 +21,7 @@ import scalacache.memcached._
 import scalacache.serialization.binary._
 import cats.effect.IO
 
-implicit val memcachedCache: Cache[IO, String] = MemcachedCache("localhost:11211")
+implicit val memcachedCache: Cache[IO, String, String] = MemcachedCache("localhost:11211")
 ```
 
 or provide your own Memcached client, like this:
@@ -36,7 +36,7 @@ val memcachedClient = new MemcachedClient(
   new BinaryConnectionFactory(), 
   AddrUtil.getAddresses("localhost:11211")
 )
-implicit val customisedMemcachedCache: Cache[IO, String] = MemcachedCache(memcachedClient)
+implicit val customisedMemcachedCache: Cache[IO, String, String] = MemcachedCache(memcachedClient)
 ```
 
 #### Keys
@@ -65,7 +65,7 @@ import scalacache.redis._
 import scalacache.serialization.binary._
 import cats.effect.IO
 
-implicit val redisCache: Cache[IO, String] = RedisCache("host1", 6379)
+implicit val redisCache: Cache[IO, String, String] = RedisCache("host1", 6379)
 ```
 
 or provide your own [Jedis](https://github.com/xetorthio/jedis) client, like this:
@@ -78,7 +78,7 @@ import _root_.redis.clients.jedis._
 import cats.effect.IO
 
 val jedisPool = new JedisPool("localhost", 6379)
-implicit val customisedRedisCache: Cache[IO, String] = RedisCache(jedisPool)
+implicit val customisedRedisCache: Cache[IO, String, String] = RedisCache(jedisPool)
 ```
 
 ScalaCache also supports [sharded Redis](https://github.com/xetorthio/jedis/wiki/AdvancedUsage#shardedjedis) and [Redis Sentinel](http://redis.io/topics/sentinel). Just create a `ShardedRedisCache` or `SentinelRedisCache` respectively.
@@ -101,7 +101,7 @@ import cats.effect.unsafe.implicits.global
 
 implicit val clock: Clock[IO] = Clock[IO]
 
-implicit val caffeineCache: Cache[IO, String] = CaffeineCache[IO, String].unsafeRunSync()
+implicit val caffeineCache: Cache[IO, String, String] = CaffeineCache[IO, String, String].unsafeRunSync()
 ```
 
 This will build a Caffeine cache with all the default settings. If you want to customize your Caffeine cache, then build it yourself and pass it to `CaffeineCache` like this:
@@ -114,7 +114,7 @@ import cats.effect.IO
 import cats.effect.unsafe.implicits.global
 
 val underlyingCaffeineCache = Caffeine.newBuilder().maximumSize(10000L).build[String, Entry[String]]
-implicit val customisedCaffeineCache: Cache[IO, String] = CaffeineCache(underlyingCaffeineCache)
+implicit val customisedCaffeineCache: Cache[IO, String, String] = CaffeineCache(underlyingCaffeineCache)
 ```
 
 ```scala mdoc:invisible
diff --git a/modules/docs/src/main/mdoc/docs/flags.md b/modules/docs/src/main/mdoc/docs/flags.md
index 88272dd9..8ed394b9 100644
--- a/modules/docs/src/main/mdoc/docs/flags.md
+++ b/modules/docs/src/main/mdoc/docs/flags.md
@@ -29,7 +29,7 @@ import cats.effect.unsafe.implicits.global
 
 final case class Cat(id: Int, name: String, colour: String)
 
-implicit val catsCache: Cache[IO, Cat] = MemcachedCache("localhost:11211")
+implicit val catsCache: Cache[IO, String, Cat] = MemcachedCache("localhost:11211")
 
 def getCatWithFlags(id: Int)(implicit flags: Flags): Cat = memoize(None) {
   // Do DB lookup here...
diff --git a/modules/docs/src/main/mdoc/docs/index.md b/modules/docs/src/main/mdoc/docs/index.md
index 5bf5a41d..8028810c 100644
--- a/modules/docs/src/main/mdoc/docs/index.md
+++ b/modules/docs/src/main/mdoc/docs/index.md
@@ -32,7 +32,7 @@ import scalacache.serialization.binary._
 
 final case class Cat(id: Int, name: String, colour: String)
 
-implicit val catsCache: Cache[IO, Cat] = MemcachedCache("localhost:11211")
+implicit val catsCache: Cache[IO, String, Cat] = MemcachedCache("localhost:11211")
 ```
 
 Note that we made the cache `implicit` so that the ScalaCache API can find it.
@@ -57,7 +57,7 @@ get("eric")
 remove("doraemon")
 
 // Flush the cache
-removeAll[Cat]
+removeAll[String, Cat]
 
 // Wrap any block with caching: if the key is not present in the cache,
 // the block will be executed and the value will be cached and returned
@@ -73,9 +73,6 @@ cachingF("benjamin")(ttl = None) {
     Cat(2, "Benjamin", "ginger")
   }
 }
-
-// You can also pass multiple parts to be combined into one key
-put("foo", 123, "bar")(ericTheCat) // Will be cached with key "foo:123:bar"
 ```
 
 ```scala mdoc:invisible
diff --git a/modules/docs/src/main/mdoc/docs/memoization.md b/modules/docs/src/main/mdoc/docs/memoization.md
index 3b7e2e81..ec9a15c9 100644
--- a/modules/docs/src/main/mdoc/docs/memoization.md
+++ b/modules/docs/src/main/mdoc/docs/memoization.md
@@ -17,7 +17,7 @@ import cats.effect.IO
 
 final case class Cat(id: Int, name: String, colour: String)
 
-implicit val catsCache: Cache[IO, Cat] = MemcachedCache("localhost:11211")
+implicit val catsCache: Cache[IO, String, Cat] = MemcachedCache("localhost:11211")
 
 def getCat(id: Int): IO[Cat] = memoize(Some(10.seconds)) {
   // Retrieve data from a remote API here ...
diff --git a/modules/memcached/src/main/scala/scalacache/memcached/MemcachedCache.scala b/modules/memcached/src/main/scala/scalacache/memcached/MemcachedCache.scala
index fac4a61a..befca2c7 100644
--- a/modules/memcached/src/main/scala/scalacache/memcached/MemcachedCache.scala
+++ b/modules/memcached/src/main/scala/scalacache/memcached/MemcachedCache.scala
@@ -1,18 +1,17 @@
 package scalacache.memcached
 
+import cats.effect.Async
 import net.spy.memcached.internal.{GetCompletionListener, GetFuture, OperationCompletionListener, OperationFuture}
 import net.spy.memcached.ops.StatusCode
 import net.spy.memcached.{AddrUtil, BinaryConnectionFactory, MemcachedClient}
-
+import scalacache.AbstractCache
 import scalacache.logging.Logger
-import scalacache.serialization.Codec
-import scalacache.{AbstractCache, CacheConfig}
+import scalacache.serialization.binary.BinaryCodec
+
 import scala.concurrent.duration.Duration
-import scala.util.Success
 import scala.language.higherKinds
+import scala.util.Success
 import scala.util.control.NonFatal
-import cats.effect.Async
-import cats.effect.Sync
 
 class MemcachedException(message: String) extends Exception(message)
 
@@ -21,8 +20,8 @@ class MemcachedException(message: String) extends Exception(message)
 class MemcachedCache[F[_]: Async, V](
     val client: MemcachedClient,
     val keySanitizer: MemcachedKeySanitizer = ReplaceAndTruncateSanitizer()
-)(implicit val config: CacheConfig, codec: Codec[V])
-    extends AbstractCache[F, V]
+)(implicit val codec: BinaryCodec[V])
+    extends AbstractCache[F, String, V]
     with MemcachedTTLConverter {
 
   protected def F: Async[F] = Async[F]
@@ -109,7 +108,7 @@ object MemcachedCache {
 
   /** Create a Memcached client connecting to localhost:11211 and use it for caching
     */
-  def apply[F[_]: Async, V](implicit config: CacheConfig, codec: Codec[V]): MemcachedCache[F, V] =
+  def apply[F[_]: Async, V](implicit codec: BinaryCodec[V]): MemcachedCache[F, V] =
     apply("localhost:11211")
 
   /** Create a Memcached client connecting to the given host(s) and use it for caching
@@ -119,7 +118,7 @@ object MemcachedCache {
     */
   def apply[F[_]: Async, V](
       addressString: String
-  )(implicit config: CacheConfig, codec: Codec[V]): MemcachedCache[F, V] =
+  )(implicit codec: BinaryCodec[V]): MemcachedCache[F, V] =
     apply(new MemcachedClient(new BinaryConnectionFactory(), AddrUtil.getAddresses(addressString)))
 
   /** Create a cache that uses the given Memcached client
@@ -129,7 +128,7 @@ object MemcachedCache {
     */
   def apply[F[_]: Async, V](
       client: MemcachedClient
-  )(implicit config: CacheConfig, codec: Codec[V]): MemcachedCache[F, V] =
+  )(implicit codec: BinaryCodec[V]): MemcachedCache[F, V] =
     new MemcachedCache[F, V](client)
 
 }
diff --git a/modules/memcached/src/test/scala/scalacache/memcached/MemcachedCacheSpec.scala b/modules/memcached/src/test/scala/scalacache/memcached/MemcachedCacheSpec.scala
index 1565eabf..5396c799 100644
--- a/modules/memcached/src/test/scala/scalacache/memcached/MemcachedCacheSpec.scala
+++ b/modules/memcached/src/test/scala/scalacache/memcached/MemcachedCacheSpec.scala
@@ -1,17 +1,16 @@
 package scalacache.memcached
 
-import org.scalatest._
-import net.spy.memcached._
-import scala.concurrent.duration._
-import org.scalatest.concurrent.{ScalaFutures, Eventually, IntegrationPatience}
-import org.scalatest.time.{Span, Seconds}
-
-import scala.language.postfixOps
-import scalacache.serialization.Codec
-import scalacache.serialization.binary._
 import cats.effect.IO
+import net.spy.memcached._
+import org.scalatest._
+import org.scalatest.concurrent.{Eventually, IntegrationPatience, ScalaFutures}
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
+import org.scalatest.time.{Seconds, Span}
+import scalacache.serialization.binary._
+
+import scala.concurrent.duration._
+import scala.language.postfixOps
 
 class MemcachedCacheSpec
     extends AnyFlatSpec
@@ -37,7 +36,7 @@ class MemcachedCacheSpec
     } catch { case _: Exception => false }
   }
 
-  def serialise[A](v: A)(implicit codec: Codec[A]): Array[Byte] =
+  def serialise[A](v: A)(implicit codec: BinaryCodec[A]): Array[Byte] =
     codec.encode(v)
 
   if (!memcachedIsRunning) {
diff --git a/modules/redis/src/main/scala/scalacache/redis/RedisCache.scala b/modules/redis/src/main/scala/scalacache/redis/RedisCache.scala
index 358d7940..eeb1b510 100644
--- a/modules/redis/src/main/scala/scalacache/redis/RedisCache.scala
+++ b/modules/redis/src/main/scala/scalacache/redis/RedisCache.scala
@@ -1,18 +1,17 @@
 package scalacache.redis
 
+import cats.effect.{MonadCancel, Sync}
 import redis.clients.jedis._
+import scalacache.serialization.binary.{BinaryCodec, BinaryEncoder}
 
 import scala.language.higherKinds
-import scalacache.CacheConfig
-import scalacache.serialization.Codec
-import cats.effect.{MonadCancel, Resource, Sync}
 
 /** Thin wrapper around Jedis
   */
-class RedisCache[F[_]: Sync, V](val jedisPool: JedisPool)(implicit
-    val config: CacheConfig,
-    val codec: Codec[V]
-) extends RedisCacheBase[F, V] {
+class RedisCache[F[_]: Sync, K, V](val jedisPool: JedisPool)(implicit
+    val keyEncoder: BinaryEncoder[K],
+    val codec: BinaryCodec[V]
+) extends RedisCacheBase[F, K, V] {
 
   protected def F: Sync[F] = Sync[F]
   type JClient = Jedis
@@ -26,19 +25,19 @@ object RedisCache {
 
   /** Create a Redis client connecting to the given host and use it for caching
     */
-  def apply[F[_]: Sync, V](
+  def apply[F[_]: Sync, K, V](
       host: String,
       port: Int
-  )(implicit config: CacheConfig, codec: Codec[V]): RedisCache[F, V] =
+  )(implicit keyEncoder: BinaryEncoder[K], codec: BinaryCodec[V]): RedisCache[F, K, V] =
     apply(new JedisPool(host, port))
 
   /** Create a cache that uses the given Jedis client pool
     * @param jedisPool
     *   a Jedis pool
     */
-  def apply[F[_]: Sync, V](
+  def apply[F[_]: Sync, K, V](
       jedisPool: JedisPool
-  )(implicit config: CacheConfig, codec: Codec[V]): RedisCache[F, V] =
-    new RedisCache[F, V](jedisPool)
+  )(implicit keyEncoder: BinaryEncoder[K], codec: BinaryCodec[V]): RedisCache[F, K, V] =
+    new RedisCache[F, K, V](jedisPool)
 
 }
diff --git a/modules/redis/src/main/scala/scalacache/redis/RedisCacheBase.scala b/modules/redis/src/main/scala/scalacache/redis/RedisCacheBase.scala
index ec645edf..57421337 100644
--- a/modules/redis/src/main/scala/scalacache/redis/RedisCacheBase.scala
+++ b/modules/redis/src/main/scala/scalacache/redis/RedisCacheBase.scala
@@ -7,32 +7,31 @@ import java.io.Closeable
 
 import scalacache.logging.Logger
 import scalacache.serialization.Codec
-import scalacache.{AbstractCache, CacheConfig}
+import scalacache.serialization.binary.{BinaryCodec, BinaryEncoder}
 
+import java.io.Closeable
 import scala.concurrent.duration._
 import cats.effect.Resource
 import cats.syntax.all._
+import scalacache.AbstractCache
 
 /** Contains implementations of all methods that can be implemented independent of the type of Redis client. This is
   * everything apart from `removeAll`, which needs to be implemented differently for sharded Redis.
   */
-trait RedisCacheBase[F[_], V] extends AbstractCache[F, V] {
+trait RedisCacheBase[F[_], K, V] extends AbstractCache[F, K, V] {
 
   override protected final val logger = Logger.getLogger[F](getClass.getName)
 
-  import StringEnrichment.StringWithUtf8Bytes
-
-  def config: CacheConfig
-
   protected type JClient <: BinaryJedisCommands with Closeable
 
   protected def jedisPool: Pool[JClient]
 
-  protected def codec: Codec[V]
+  protected def keyEncoder: BinaryEncoder[K]
+  protected def codec: BinaryCodec[V]
 
-  protected def doGet(key: String): F[Option[V]] =
+  protected def doGet(key: K): F[Option[V]] =
     withJedis { jedis =>
-      val bytes = jedis.get(key.utf8bytes)
+      val bytes = jedis.get(keyEncoder.encode(key))
       val result: Codec.DecodingResult[Option[V]] = {
         if (bytes != null)
           codec.decode(bytes).right.map(Some(_))
@@ -48,9 +47,9 @@ trait RedisCacheBase[F[_], V] extends AbstractCache[F, V] {
       }
     }
 
-  protected def doPut(key: String, value: V, ttl: Option[Duration]): F[Unit] = {
+  protected def doPut(key: K, value: V, ttl: Option[Duration]): F[Unit] = {
     withJedis { jedis =>
-      val keyBytes   = key.utf8bytes
+      val keyBytes   = keyEncoder.encode(key)
       val valueBytes = codec.encode(value)
       ttl match {
         case None                => F.delay(jedis.set(keyBytes, valueBytes))
@@ -69,9 +68,9 @@ trait RedisCacheBase[F[_], V] extends AbstractCache[F, V] {
     } *> logCachePut(key, ttl)
   }
 
-  protected def doRemove(key: String): F[Unit] = {
+  protected def doRemove(key: K): F[Unit] = {
     withJedis { jedis =>
-      F.delay(jedis.del(key.utf8bytes))
+      F.delay(jedis.del(keyEncoder.encode(key)))
     }
   }
 
diff --git a/modules/redis/src/main/scala/scalacache/redis/RedisClusterCache.scala b/modules/redis/src/main/scala/scalacache/redis/RedisClusterCache.scala
index 22b5c727..4b3d8c4c 100644
--- a/modules/redis/src/main/scala/scalacache/redis/RedisClusterCache.scala
+++ b/modules/redis/src/main/scala/scalacache/redis/RedisClusterCache.scala
@@ -1,27 +1,27 @@
 package scalacache.redis
 
 import scalacache.logging.Logger
-import scalacache.redis.StringEnrichment._
 import scalacache.serialization.Codec
-import scalacache.{AbstractCache, CacheConfig}
+import scalacache.serialization.binary.{BinaryCodec, BinaryEncoder}
 
 import scala.concurrent.duration.{Duration, _}
 import cats.implicits._
 import cats.effect.Sync
 import redis.clients.jedis.JedisCluster
 import redis.clients.jedis.exceptions.JedisClusterException
+import scalacache.AbstractCache
 
-class RedisClusterCache[F[_]: Sync, V](val jedisCluster: JedisCluster)(implicit
-    val config: CacheConfig,
-    val codec: Codec[V]
-) extends AbstractCache[F, V] {
+class RedisClusterCache[F[_]: Sync, K, V](val jedisCluster: JedisCluster)(implicit
+    val keyEncoder: BinaryEncoder[K],
+    val codec: BinaryCodec[V]
+) extends AbstractCache[F, K, V] {
 
   protected def F: Sync[F] = Sync[F]
 
   override protected final val logger = Logger.getLogger(getClass.getName)
 
-  override protected def doGet(key: String): F[Option[V]] = F.defer {
-    val bytes = jedisCluster.get(key.utf8bytes)
+  override protected def doGet(key: K): F[Option[V]] = F.defer {
+    val bytes = jedisCluster.get(keyEncoder.encode(key))
     val result: Codec.DecodingResult[Option[V]] = {
       if (bytes != null)
         codec.decode(bytes).right.map(Some(_))
@@ -37,8 +37,8 @@ class RedisClusterCache[F[_]: Sync, V](val jedisCluster: JedisCluster)(implicit
     }
   }
 
-  override protected def doPut(key: String, value: V, ttl: Option[Duration]): F[Unit] = {
-    val keyBytes   = key.utf8bytes
+  override protected def doPut(key: K, value: V, ttl: Option[Duration]): F[Unit] = {
+    val keyBytes   = keyEncoder.encode(key)
     val valueBytes = codec.encode(value)
     ttl match {
       case None                => F.delay(jedisCluster.set(keyBytes, valueBytes))
@@ -54,8 +54,8 @@ class RedisClusterCache[F[_]: Sync, V](val jedisCluster: JedisCluster)(implicit
     }
   }
 
-  override protected def doRemove(key: String): F[Unit] = F.delay {
-    jedisCluster.del(key.utf8bytes)
+  override protected def doRemove(key: K): F[Unit] = F.delay {
+    jedisCluster.del(keyEncoder.encode(key))
   }
 
   @deprecated(
diff --git a/modules/redis/src/main/scala/scalacache/redis/RedisSerialization.scala b/modules/redis/src/main/scala/scalacache/redis/RedisSerialization.scala
index 9443aca4..76fbe987 100644
--- a/modules/redis/src/main/scala/scalacache/redis/RedisSerialization.scala
+++ b/modules/redis/src/main/scala/scalacache/redis/RedisSerialization.scala
@@ -1,16 +1,17 @@
 package scalacache.redis
 
 import scalacache.serialization.Codec
+import scalacache.serialization.binary.BinaryCodec
 
 /** Custom serialization for caching arbitrary objects in Redis. Ints, Longs, Doubles, Strings and byte arrays are
   * treated specially. Everything else is serialized using standard Java serialization.
   */
 trait RedisSerialization {
 
-  def serialize[A](value: A)(implicit codec: Codec[A]): Array[Byte] =
+  def serialize[A](value: A)(implicit codec: BinaryCodec[A]): Array[Byte] =
     codec.encode(value)
 
-  def deserialize[A](bytes: Array[Byte])(implicit codec: Codec[A]): Codec.DecodingResult[A] =
+  def deserialize[A](bytes: Array[Byte])(implicit codec: BinaryCodec[A]): Codec.DecodingResult[A] =
     codec.decode(bytes)
 
 }
diff --git a/modules/redis/src/main/scala/scalacache/redis/SentinelRedisCache.scala b/modules/redis/src/main/scala/scalacache/redis/SentinelRedisCache.scala
index b0225622..a48750f9 100644
--- a/modules/redis/src/main/scala/scalacache/redis/SentinelRedisCache.scala
+++ b/modules/redis/src/main/scala/scalacache/redis/SentinelRedisCache.scala
@@ -1,20 +1,18 @@
 package scalacache.redis
 
+import cats.effect.{MonadCancel, Sync}
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig
 import redis.clients.jedis._
+import scalacache.serialization.binary.{BinaryCodec, BinaryEncoder}
 
 import scala.collection.JavaConverters._
-import scalacache.CacheConfig
-import scalacache.serialization.Codec
-import cats.implicits._
-import cats.effect.{MonadCancel, Sync}
-import org.apache.commons.pool2.impl.GenericObjectPoolConfig
 
 /** Thin wrapper around Jedis that works with Redis Sentinel.
   */
-class SentinelRedisCache[F[_]: Sync, V](val jedisPool: JedisSentinelPool)(implicit
-    val config: CacheConfig,
-    val codec: Codec[V]
-) extends RedisCacheBase[F, V] {
+class SentinelRedisCache[F[_]: Sync, K, V](val jedisPool: JedisSentinelPool)(implicit
+    val keyEncoder: BinaryEncoder[K],
+    val codec: BinaryCodec[V]
+) extends RedisCacheBase[F, K, V] {
 
   protected def F: Sync[F] = Sync[F]
 
@@ -38,10 +36,10 @@ object SentinelRedisCache {
     * @param password
     *   password of the cluster
     */
-  def apply[F[_]: Sync, V](clusterName: String, sentinels: Set[String], password: String)(implicit
-      config: CacheConfig,
-      codec: Codec[V]
-  ): SentinelRedisCache[F, V] =
+  def apply[F[_]: Sync, K, V](clusterName: String, sentinels: Set[String], password: String)(implicit
+      keyEncoder: BinaryEncoder[K],
+      codec: BinaryCodec[V]
+  ): SentinelRedisCache[F, K, V] =
     apply(new JedisSentinelPool(clusterName, sentinels.asJava, new GenericObjectPoolConfig[Jedis], password))
 
   /** Create a `SentinelRedisCache` that uses a `JedisSentinelPool` with a custom pool config.
@@ -55,15 +53,15 @@ object SentinelRedisCache {
     * @param poolConfig
     *   config of the underlying pool
     */
-  def apply[F[_]: Sync, V](
+  def apply[F[_]: Sync, K, V](
       clusterName: String,
       sentinels: Set[String],
       poolConfig: GenericObjectPoolConfig[Jedis],
       password: String
   )(implicit
-      config: CacheConfig,
-      codec: Codec[V]
-  ): SentinelRedisCache[F, V] =
+      keyEncoder: BinaryEncoder[K],
+      codec: BinaryCodec[V]
+  ): SentinelRedisCache[F, K, V] =
     apply(new JedisSentinelPool(clusterName, sentinels.asJava, poolConfig, password))
 
   /** Create a `SentinelRedisCache` that uses the given JedisSentinelPool
@@ -71,9 +69,9 @@ object SentinelRedisCache {
     * @param jedisSentinelPool
     *   a JedisSentinelPool
     */
-  def apply[F[_]: Sync, V](
+  def apply[F[_]: Sync, K, V](
       jedisSentinelPool: JedisSentinelPool
-  )(implicit config: CacheConfig, codec: Codec[V]): SentinelRedisCache[F, V] =
-    new SentinelRedisCache[F, V](jedisSentinelPool)
+  )(implicit keyEncoder: BinaryEncoder[K], codec: BinaryCodec[V]): SentinelRedisCache[F, K, V] =
+    new SentinelRedisCache[F, K, V](jedisSentinelPool)
 
 }
diff --git a/modules/redis/src/main/scala/scalacache/redis/ShardedRedisCache.scala b/modules/redis/src/main/scala/scalacache/redis/ShardedRedisCache.scala
index 2bdcf392..9d0f27f7 100644
--- a/modules/redis/src/main/scala/scalacache/redis/ShardedRedisCache.scala
+++ b/modules/redis/src/main/scala/scalacache/redis/ShardedRedisCache.scala
@@ -1,20 +1,19 @@
 package scalacache.redis
 
+import cats.effect.Sync
 import redis.clients.jedis._
+import scalacache.serialization.binary.{BinaryCodec, BinaryEncoder}
+import org.apache.commons.pool2.impl.GenericObjectPoolConfig
 
 import scala.collection.JavaConverters._
 import scala.language.higherKinds
-import scalacache.CacheConfig
-import scalacache.serialization.Codec
-import cats.effect.{MonadCancel, Sync}
-import org.apache.commons.pool2.impl.GenericObjectPoolConfig
 
 /** Thin wrapper around Jedis that works with sharded Redis.
   */
-class ShardedRedisCache[F[_]: Sync, V](val jedisPool: ShardedJedisPool)(implicit
-    val config: CacheConfig,
-    val codec: Codec[V]
-) extends RedisCacheBase[F, V] {
+class ShardedRedisCache[F[_]: Sync, K, V](val jedisPool: ShardedJedisPool)(implicit
+    val keyEncoder: BinaryEncoder[K],
+    val codec: BinaryCodec[V]
+) extends RedisCacheBase[F, K, V] {
 
   protected def F: Sync[F] = Sync[F]
 
@@ -32,9 +31,9 @@ object ShardedRedisCache {
 
   /** Create a sharded Redis client connecting to the given hosts and use it for caching
     */
-  def apply[F[_]: Sync, V](
+  def apply[F[_]: Sync, K, V](
       hosts: (String, Int)*
-  )(implicit config: CacheConfig, codec: Codec[V]): ShardedRedisCache[F, V] = {
+  )(implicit keyEncoder: BinaryEncoder[K], codec: BinaryCodec[V]): ShardedRedisCache[F, K, V] = {
     val shards = hosts.map { case (host, port) =>
       new JedisShardInfo(host, port)
     }
@@ -46,9 +45,9 @@ object ShardedRedisCache {
     * @param jedisPool
     *   a ShardedJedis pool
     */
-  def apply[F[_]: Sync, V](
+  def apply[F[_]: Sync, K, V](
       jedisPool: ShardedJedisPool
-  )(implicit config: CacheConfig, codec: Codec[V]): ShardedRedisCache[F, V] =
-    new ShardedRedisCache[F, V](jedisPool)
+  )(implicit keyEncoder: BinaryEncoder[K], codec: BinaryCodec[V]): ShardedRedisCache[F, K, V] =
+    new ShardedRedisCache[F, K, V](jedisPool)
 
 }
diff --git a/modules/redis/src/main/scala/scalacache/redis/StringEnrichment.scala b/modules/redis/src/main/scala/scalacache/redis/StringEnrichment.scala
deleted file mode 100644
index a1db1b8c..00000000
--- a/modules/redis/src/main/scala/scalacache/redis/StringEnrichment.scala
+++ /dev/null
@@ -1,15 +0,0 @@
-package scalacache.redis
-
-import java.nio.charset.Charset
-
-object StringEnrichment {
-
-  private val utf8 = Charset.forName("UTF-8")
-
-  /** Enrichment class to convert String to UTF-8 byte array
-    */
-  implicit class StringWithUtf8Bytes(val string: String) extends AnyVal {
-    def utf8bytes = string.getBytes(utf8)
-  }
-
-}
diff --git a/modules/redis/src/test/scala/scalacache/redis/Issue32Spec.scala b/modules/redis/src/test/scala/scalacache/redis/Issue32Spec.scala
index 9baaf0d8..a138c759 100644
--- a/modules/redis/src/test/scala/scalacache/redis/Issue32Spec.scala
+++ b/modules/redis/src/test/scala/scalacache/redis/Issue32Spec.scala
@@ -1,14 +1,14 @@
 package scalacache.redis
 
 import org.scalatest.BeforeAndAfter
-
-import scalacache.Cache
 import scalacache.memoization._
 import scalacache.serialization.binary._
 import cats.effect.IO
 import cats.effect.unsafe.implicits.global
 import org.scalatest.flatspec.AnyFlatSpec
 import org.scalatest.matchers.should.Matchers
+import scalacache.memoization.MemoizationConfig.defaultMemoizationConfig
+import scalacache.serialization.binary.StringBinaryCodec
 
 case class User(id: Int, name: String)
 
@@ -18,7 +18,7 @@ case class User(id: Int, name: String)
 class Issue32Spec extends AnyFlatSpec with Matchers with BeforeAndAfter with RedisTestUtil {
 
   assumingRedisIsRunning { (pool, client) =>
-    implicit val cache: Cache[IO, List[User]] = RedisCache[IO, List[User]](pool)
+    implicit val cache: RedisCache[IO, String, List[User]] = new RedisCache[IO, String, List[User]](pool)
 
     def getUser(id: Int): List[User] =
       memoize(None) {
diff --git a/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpec.scala b/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpec.scala
index 4709a4df..2cca55da 100644
--- a/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpec.scala
+++ b/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpec.scala
@@ -6,6 +6,8 @@ import scalacache.serialization.Codec
 
 import scala.language.postfixOps
 import cats.effect.IO
+import scalacache.serialization.binary.BinaryCodec
+import scalacache.serialization.binary.StringBinaryCodec
 
 class RedisCacheSpec extends RedisCacheSpecBase with RedisTestUtil {
 
@@ -14,8 +16,8 @@ class RedisCacheSpec extends RedisCacheSpecBase with RedisTestUtil {
 
   val withJedis = assumingRedisIsRunning _
 
-  def constructCache[V](pool: JPool)(implicit codec: Codec[V]): CacheAlg[IO, V] =
-    new RedisCache[IO, V](jedisPool = pool)
+  def constructCache[V](pool: JPool)(implicit codec: BinaryCodec[V]): Cache[IO, String, V] =
+    new RedisCache[IO, String, V](jedisPool = pool)
 
   def flushRedis(client: JClient): Unit = client.underlying.flushDB()
 
diff --git a/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpecBase.scala b/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpecBase.scala
index da809c22..01f44165 100644
--- a/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpecBase.scala
+++ b/modules/redis/src/test/scala/scalacache/redis/RedisCacheSpecBase.scala
@@ -30,14 +30,14 @@ trait RedisCacheSpecBase
   type JClient <: BaseJedisClient
 
   case object AlwaysFailing
-  implicit val alwaysFailingCodec: Codec[AlwaysFailing.type] = new Codec[AlwaysFailing.type] {
+  implicit val alwaysFailingCodec: BinaryCodec[AlwaysFailing.type] = new BinaryCodec[AlwaysFailing.type] {
     override def encode(value: AlwaysFailing.type): Array[Byte] = Array(0)
     override def decode(bytes: Array[Byte]): DecodingResult[AlwaysFailing.type] =
       Left(FailedToDecode(new Exception("Failed to decode")))
   }
 
   def withJedis: ((JPool, JClient) => Unit) => Unit
-  def constructCache[V](pool: JPool)(implicit codec: Codec[V]): CacheAlg[IO, V]
+  def constructCache[V](pool: JPool)(implicit codec: BinaryCodec[V]): Cache[IO, String, V]
   def flushRedis(client: JClient): Unit
 
   def runTestsIfPossible() = {
@@ -114,7 +114,7 @@ trait RedisCacheSpecBase
 
       behavior of "caching with serialization"
 
-      def roundTrip[V](key: String, value: V)(implicit codec: Codec[V]): Future[Option[V]] = {
+      def roundTrip[V](key: String, value: V)(implicit codec: BinaryCodec[V]): Future[Option[V]] = {
         val c = constructCache[V](pool)
         c.put(key)(value, None).flatMap(_ => c.get(key)).unsafeToFuture()
       }
diff --git a/modules/redis/src/test/scala/scalacache/redis/RedisClusterCacheSpec.scala b/modules/redis/src/test/scala/scalacache/redis/RedisClusterCacheSpec.scala
index 5bc94ccf..3cff05da 100644
--- a/modules/redis/src/test/scala/scalacache/redis/RedisClusterCacheSpec.scala
+++ b/modules/redis/src/test/scala/scalacache/redis/RedisClusterCacheSpec.scala
@@ -8,6 +8,8 @@ import scala.collection.JavaConverters._
 import scala.language.postfixOps
 import scala.util.{Failure, Success, Try}
 import cats.effect.IO
+import scalacache.serialization.binary.BinaryCodec
+import scalacache.serialization.binary.StringBinaryCodec
 
 class RedisClusterCacheSpec extends RedisCacheSpecBase with RedisTestUtil {
 
@@ -16,8 +18,8 @@ class RedisClusterCacheSpec extends RedisCacheSpecBase with RedisTestUtil {
 
   override val withJedis = assumingRedisClusterIsRunning _
 
-  def constructCache[V](jedisCluster: JedisCluster)(implicit codec: Codec[V]): CacheAlg[IO, V] =
-    new RedisClusterCache[IO, V](jedisCluster)
+  def constructCache[V](jedisCluster: JedisCluster)(implicit codec: BinaryCodec[V]): Cache[IO, String, V] =
+    new RedisClusterCache[IO, String, V](jedisCluster)
 
   def flushRedis(client: JClient): Unit =
     client.underlying.getClusterNodes.asScala.mapValues(_.getResource.flushDB())
diff --git a/modules/redis/src/test/scala/scalacache/redis/SentinelRedisCacheSpec.scala b/modules/redis/src/test/scala/scalacache/redis/SentinelRedisCacheSpec.scala
index 3ca99195..7a4e115b 100644
--- a/modules/redis/src/test/scala/scalacache/redis/SentinelRedisCacheSpec.scala
+++ b/modules/redis/src/test/scala/scalacache/redis/SentinelRedisCacheSpec.scala
@@ -8,6 +8,8 @@ import scala.util.{Failure, Success, Try}
 import scalacache._
 import scalacache.serialization.Codec
 import cats.effect.IO
+import scalacache.serialization.binary.BinaryCodec
+import scalacache.serialization.binary.StringBinaryCodec
 
 class SentinelRedisCacheSpec extends RedisCacheSpecBase {
 
@@ -16,8 +18,8 @@ class SentinelRedisCacheSpec extends RedisCacheSpecBase {
 
   val withJedis = assumingRedisSentinelIsRunning _
 
-  def constructCache[V](pool: JPool)(implicit codec: Codec[V]): CacheAlg[IO, V] =
-    new SentinelRedisCache[IO, V](jedisPool = pool)
+  def constructCache[V](pool: JPool)(implicit codec: BinaryCodec[V]): Cache[IO, String, V] =
+    new SentinelRedisCache[IO, String, V](jedisPool = pool)
 
   def flushRedis(client: JClient): Unit = client.underlying.flushDB()
 
diff --git a/modules/redis/src/test/scala/scalacache/redis/ShardedRedisCacheSpec.scala b/modules/redis/src/test/scala/scalacache/redis/ShardedRedisCacheSpec.scala
index c86dabbb..090151b3 100644
--- a/modules/redis/src/test/scala/scalacache/redis/ShardedRedisCacheSpec.scala
+++ b/modules/redis/src/test/scala/scalacache/redis/ShardedRedisCacheSpec.scala
@@ -7,7 +7,9 @@ import scalacache.serialization.Codec
 import scala.collection.JavaConverters._
 import scala.util.{Failure, Success, Try}
 import cats.effect.IO
+import scalacache.serialization.binary.BinaryCodec
 import org.apache.commons.pool2.impl.GenericObjectPoolConfig
+import scalacache.serialization.binary.StringBinaryCodec
 
 class ShardedRedisCacheSpec extends RedisCacheSpecBase {
 
@@ -16,8 +18,8 @@ class ShardedRedisCacheSpec extends RedisCacheSpecBase {
 
   val withJedis = assumingMultipleRedisAreRunning _
 
-  def constructCache[V](pool: JPool)(implicit codec: Codec[V]): CacheAlg[IO, V] =
-    new ShardedRedisCache[IO, V](jedisPool = pool)
+  def constructCache[V](pool: JPool)(implicit codec: BinaryCodec[V]): Cache[IO, String, V] =
+    new ShardedRedisCache[IO, String, V](jedisPool = pool)
 
   def flushRedis(client: JClient): Unit =
     client.underlying.getAllShards.asScala.foreach(_.flushDB())
diff --git a/modules/tests/src/test/scala/integrationtests/IntegrationTests.scala b/modules/tests/src/test/scala/integrationtests/IntegrationTests.scala
index 5d764f2c..da9eb762 100644
--- a/modules/tests/src/test/scala/integrationtests/IntegrationTests.scala
+++ b/modules/tests/src/test/scala/integrationtests/IntegrationTests.scala
@@ -51,9 +51,9 @@ class IntegrationTests extends AnyFlatSpec with Matchers with BeforeAndAfterAll
     }
   }
 
-  case class CacheBackend(name: String, cache: Cache[CatsIO, String])
+  case class CacheBackend(name: String, cache: Cache[CatsIO, String, String])
 
-  private val caffeine = CacheBackend("Caffeine", CaffeineCache[CatsIO, String].unsafeRunSync())
+  private val caffeine = CacheBackend("Caffeine", CaffeineCache[CatsIO, String, String].unsafeRunSync())
   private val memcached: Seq[CacheBackend] =
     if (memcachedIsRunning) {
       Seq(
@@ -75,10 +75,10 @@ class IntegrationTests extends AnyFlatSpec with Matchers with BeforeAndAfterAll
       Seq(
         {
           import scalacache.serialization.binary._
-          CacheBackend("(Redis) ⇔ (binary codec)", RedisCache[CatsIO, String](jedisPool))
+          CacheBackend("(Redis) ⇔ (binary codec)", RedisCache[CatsIO, String, String](jedisPool))
         }, {
           import scalacache.serialization.circe._
-          CacheBackend("(Redis) ⇔ (circe codec)", RedisCache[CatsIO, String](jedisPool))
+          CacheBackend("(Redis) ⇔ (circe codec)", RedisCache[CatsIO, String, String](jedisPool))
         }
       )
     else {
@@ -91,7 +91,7 @@ class IntegrationTests extends AnyFlatSpec with Matchers with BeforeAndAfterAll
   for (CacheBackend(name, cache) <- backends) {
 
     s"$name ⇔ (cats-effect IO)" should "defer the computation and give the correct result" in {
-      implicit val theCache: Cache[CatsIO, String] = cache
+      implicit val theCache: Cache[CatsIO, String, String] = cache
 
       val key          = UUID.randomUUID().toString
       val initialValue = UUID.randomUUID().toString
@@ -113,7 +113,7 @@ class IntegrationTests extends AnyFlatSpec with Matchers with BeforeAndAfterAll
     }
   }
 
-  private def checkComputationHasNotRun(key: String)(implicit cache: Cache[CatsIO, String]): Unit = {
+  private def checkComputationHasNotRun(key: String)(implicit cache: Cache[CatsIO, String, String]): Unit = {
     Thread.sleep(1000)
     assert(cache.get(key).unsafeRunSync().isEmpty)
   }