diff --git a/.idea/misc.xml b/.idea/misc.xml
index 8dc554d..206c814 100644
--- a/.idea/misc.xml
+++ b/.idea/misc.xml
@@ -8,7 +8,7 @@
       </list>
     </option>
   </component>
-  <component name="ProjectRootManager" version="2" project-jdk-name="openjdk-22" project-jdk-type="JavaSDK" />
+  <component name="ProjectRootManager" version="2" languageLevel="JDK_23" default="true" project-jdk-name="corretto-23" project-jdk-type="JavaSDK" />
   <component name="ProjectType">
     <option name="id" value="jpab" />
   </component>
diff --git a/pom.xml b/pom.xml
index b4c0aac..8f2591b 100644
--- a/pom.xml
+++ b/pom.xml
@@ -20,6 +20,21 @@
             <artifactId>jedis</artifactId>
             <version>5.1.5</version>
         </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+            <version>2.17.1</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-core</artifactId>
+            <version>2.17.1</version>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <version>2.17.1</version>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>
diff --git a/src/main/java/org/redis/map/RedisMap.java b/src/main/java/org/redis/map/RedisMap.java
new file mode 100644
index 0000000..d970c13
--- /dev/null
+++ b/src/main/java/org/redis/map/RedisMap.java
@@ -0,0 +1,207 @@
+package org.redis.map;
+
+import org.redis.scan.ScanIterator;
+import org.redis.serializer.Serializer;
+import redis.clients.jedis.Jedis;
+import redis.clients.jedis.JedisPool;
+import redis.clients.jedis.resps.ScanResult;
+
+import java.util.*;
+
+public class RedisMap<K, V> implements Map<K, V> {
+
+    private final JedisPool pool;
+    private final String redisKey;
+    private final Serializer<K> keySerializer;
+    private final Serializer<V> valueSerializer;
+    private final Class<K> keyClass;
+    private final Class<V> valueClass;
+
+    public RedisMap(JedisPool pool,
+                    String redisKey,
+                    Serializer<K> keySerializer,
+                    Serializer<V> valueSerializer,
+                    Class<K> keyClass,
+                    Class<V> valueClass) {
+        this.pool = Objects.requireNonNull(pool, "pool must not be null");
+        this.redisKey = Objects.requireNonNull(redisKey, "redisKey must not be null");
+        this.keySerializer = Objects.requireNonNull(keySerializer, "keySerializer must not be null");
+        this.valueSerializer = Objects.requireNonNull(valueSerializer, "valueSerializer must not be null");
+        this.keyClass = Objects.requireNonNull(keyClass, "keyClass must not be null");
+        this.valueClass = Objects.requireNonNull(valueClass, "valueClass must not be null");
+    }
+
+    private String serializeKey(K key) {
+        Objects.requireNonNull(key, "key must not be null");
+        return keySerializer.serialize(key);
+    }
+
+    private String serializeValue(V value) {
+        Objects.requireNonNull(value, "value must not be null");
+        return valueSerializer.serialize(value);
+    }
+
+    private K deserializeKey(String raw) {
+        return keySerializer.deserialize(raw);
+    }
+
+    private V deserializeValue(String raw) {
+        return valueSerializer.deserialize(raw);
+    }
+
+    @Override
+    public int size() {
+        try (Jedis jedis = pool.getResource()) {
+            long len = jedis.hlen(redisKey);
+            if (len > Integer.MAX_VALUE) {
+                throw new IllegalStateException("Redis hash size exceed Integer.MAX_VALUE: " + len);
+            }
+            return (int) len;
+        }
+    }
+
+    @Override
+    public boolean isEmpty() {
+        return size() == 0;
+    }
+
+    @Override
+    public boolean containsKey(Object key) {
+        if (!keyClass.isInstance(key)) return false;
+        K castedKey = keyClass.cast(key);
+        try (Jedis jedis = pool.getResource()) {
+            return jedis.hexists(redisKey, serializeKey(castedKey));
+        }
+    }
+
+    @Override
+    public boolean containsValue(Object value) {
+        if (!valueClass.isInstance(value)) return false;
+        V castedValue = valueClass.cast(value);
+        Iterator<V> it = new ScanIterator<>(this::scanValues);
+        while (it.hasNext()) {
+            if (Objects.equals(it.next(), castedValue)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    @Override
+    public V get(Object key) {
+        if (!keyClass.isInstance(key)) return null;
+        K castedKey = keyClass.cast(key);
+        try (Jedis jedis = pool.getResource()) {
+            String raw = jedis.hget(redisKey, serializeKey(castedKey));
+            return raw != null ? deserializeValue(raw) : null;
+        }
+    }
+
+    @Override
+    public V put(K key, V value) {
+        String serializeKey = serializeKey(key);
+        String serializeValue = serializeValue(value);
+        try (Jedis jedis = pool.getResource()) {
+            String prev = jedis.hget(redisKey, serializeKey);
+            jedis.hset(redisKey, serializeKey, serializeValue);
+            return prev != null ? deserializeValue(prev) : null;
+        }
+    }
+
+    @Override
+    public V remove(Object key) {
+        if (!keyClass.isInstance(key)) return null;
+        K castedKey = keyClass.cast(key);
+        String serializeKey = serializeKey(castedKey);
+        try (Jedis jedis = pool.getResource()) {
+            String prev = jedis.hget(redisKey, serializeKey);
+            if (prev != null) jedis.hdel(redisKey, serializeKey);
+            return prev != null ? deserializeValue(prev) : null;
+        }
+    }
+
+    @Override
+    public void putAll(Map<? extends K, ? extends V> m) {
+        if (m == null || m.isEmpty()) return;
+        Map<String, String> bulk = new HashMap<>(m.size());
+        for (Entry<? extends K, ? extends V> e : m.entrySet()) {
+            bulk.put(serializeKey(e.getKey()), serializeValue(e.getValue()));
+        }
+        try (Jedis jedis = pool.getResource()) {
+            jedis.hset(redisKey, bulk);
+        }
+    }
+
+    @Override
+    public void clear() {
+        try (Jedis jedis = pool.getResource()) {
+            jedis.del(redisKey);
+        }
+    }
+
+    @Override
+    public Set<K> keySet() {
+        Set<K> keys = new HashSet<>();
+        Iterator<K> it = new ScanIterator<>(this::scanKeys);
+        while (it.hasNext()) {
+            keys.add(it.next());
+        }
+        return keys;
+    }
+
+    @Override
+    public Collection<V> values() {
+        List<V> values = new ArrayList<>();
+        Iterator<V> it = new ScanIterator<>(this::scanValues);
+        while (it.hasNext()) {
+            values.add(it.next());
+        }
+        return values;
+    }
+
+    @Override
+    public Set<Entry<K, V>> entrySet() {
+        Set<Entry<K, V>> entries = new HashSet<>();
+        Iterator<Entry<K, V>> it = new ScanIterator<>(this::scanEntries);
+        while (it.hasNext()) {
+            entries.add(it.next());
+        }
+        return entries;
+    }
+
+    private ScanResult<K> scanKeys(String cursor) {
+        try (Jedis jedis = pool.getResource()) {
+            ScanResult<Map.Entry<String, String>> result = jedis.hscan(redisKey, cursor);
+            List<K> keys = new ArrayList<>(result.getResult().size());
+            for (Map.Entry<String, String> e : result.getResult()) {
+                keys.add(deserializeKey(e.getKey()));
+            }
+            return new ScanResult<>(result.getCursor(), keys);
+        }
+    }
+
+    private ScanResult<V> scanValues(String cursor) {
+        try (Jedis jedis = pool.getResource()) {
+            ScanResult<Map.Entry<String, String>> result = jedis.hscan(redisKey, cursor);
+            List<V> vals = new ArrayList<>(result.getResult().size());
+            for (Map.Entry<String, String> e : result.getResult()) {
+                vals.add(deserializeValue(e.getValue()));
+            }
+            return new ScanResult<>(result.getCursor(), vals);
+        }
+    }
+
+    private ScanResult<Entry<K, V>> scanEntries(String cursor) {
+        try (Jedis jedis = pool.getResource()) {
+            ScanResult<Map.Entry<String, String>> result = jedis.hscan(redisKey, cursor);
+            List<Entry<K, V>> list = new ArrayList<>(result.getResult().size());
+            for (Map.Entry<String, String> e : result.getResult()) {
+                list.add(new AbstractMap.SimpleEntry<>(
+                        deserializeKey(e.getKey()),
+                        deserializeValue(e.getValue())
+                ));
+            }
+            return new ScanResult<>(result.getCursor(), list);
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/redis/scan/ScanIterator.java b/src/main/java/org/redis/scan/ScanIterator.java
new file mode 100644
index 0000000..daef7e1
--- /dev/null
+++ b/src/main/java/org/redis/scan/ScanIterator.java
@@ -0,0 +1,48 @@
+package org.redis.scan;
+
+import redis.clients.jedis.params.ScanParams;
+import redis.clients.jedis.resps.ScanResult;
+
+import java.util.Collections;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+import java.util.function.Function;
+
+public class ScanIterator<T> implements Iterator<T> {
+
+    private String cursor = ScanParams.SCAN_POINTER_START;
+
+    private Iterator<T> buffer = Collections.emptyIterator();
+
+    private final Function<String, ScanResult<T>> scanner;
+
+    private boolean finished = false;
+
+
+    public ScanIterator(Function<String, ScanResult<T>> scanner) {
+        this.scanner = scanner;
+    }
+
+    @Override
+    public boolean hasNext() {
+        if (finished && !buffer.hasNext()) {
+            return false;
+        }
+
+        while (!buffer.hasNext() && !finished) {
+            ScanResult<T> result = scanner.apply(cursor);
+            cursor = result.getCursor();
+            buffer = result.getResult().iterator();
+            if ("0".equals(cursor)) {
+                finished = true;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    public T next() {
+        if (!hasNext()) throw new NoSuchElementException();
+        return buffer.next();
+    }
+}
\ No newline at end of file
diff --git a/src/main/java/org/redis/serializer/JacksonSerializer.java b/src/main/java/org/redis/serializer/JacksonSerializer.java
new file mode 100644
index 0000000..5d6642b
--- /dev/null
+++ b/src/main/java/org/redis/serializer/JacksonSerializer.java
@@ -0,0 +1,32 @@
+package org.redis.serializer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+
+public class JacksonSerializer<T> implements Serializer<T> {
+
+    private final Class<T> type;
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    public JacksonSerializer(Class<T> type) {
+        this.type = type;
+    }
+
+    @Override
+    public String serialize(T value) {
+        try {
+            return objectMapper.writeValueAsString(value);
+        } catch (Exception e) {
+            throw new IllegalArgumentException("Serialize error", e);
+        }
+    }
+
+    @Override
+    public T deserialize(String string) {
+        try {
+            return objectMapper.readValue(string, type);
+        } catch (Exception e) {
+            throw new IllegalArgumentException("Deserialize error", e);
+        }
+    }
+}
diff --git a/src/main/java/org/redis/serializer/Serializer.java b/src/main/java/org/redis/serializer/Serializer.java
new file mode 100644
index 0000000..3c70bd6
--- /dev/null
+++ b/src/main/java/org/redis/serializer/Serializer.java
@@ -0,0 +1,9 @@
+package org.redis.serializer;
+
+public interface Serializer<T> {
+
+    String serialize(T value);
+
+    T deserialize(String string);
+}
+
diff --git a/src/main/java/org/redis/serializer/StringSerializer.java b/src/main/java/org/redis/serializer/StringSerializer.java
new file mode 100644
index 0000000..48302e0
--- /dev/null
+++ b/src/main/java/org/redis/serializer/StringSerializer.java
@@ -0,0 +1,13 @@
+package org.redis.serializer;
+
+public class StringSerializer implements Serializer<String> {
+    @Override
+    public String serialize(String value) {
+        return value;
+    }
+
+    @Override
+    public String deserialize(String string) {
+        return string;
+    }
+}
diff --git a/src/test/java/org/redis/RedisMapTest.java b/src/test/java/org/redis/RedisMapTest.java
new file mode 100644
index 0000000..25207cb
--- /dev/null
+++ b/src/test/java/org/redis/RedisMapTest.java
@@ -0,0 +1,102 @@
+package org.redis;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.redis.map.RedisMap;
+import org.redis.serializer.StringSerializer;
+import redis.clients.jedis.JedisPool;
+
+import java.util.AbstractMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class RedisMapTest {
+
+    private JedisPool pool;
+    private RedisMap<String, String> map;
+
+    @BeforeEach
+    void setUp() {
+        pool = new JedisPool("localhost", 6379);
+        map = new RedisMap<>(
+                pool,
+                "test-map",
+                new StringSerializer(),
+                new StringSerializer(),
+                String.class,
+                String.class
+        );
+        map.clear();
+        map.put("a", "1");
+        map.put("b", "2");
+    }
+
+    @Test
+    void testSize() {
+        assertEquals(2, map.size());
+        assertFalse(map.isEmpty());
+    }
+
+    @Test
+    void testPutAndGet() {
+        assertEquals("1", map.get("a"));
+        assertEquals("2", map.get("b"));
+        assertEquals("1", map.put("a", "X"));
+        assertEquals("X", map.get("a"));
+    }
+
+    @Test
+    void testRemove() {
+        assertEquals("1", map.remove("a"));
+        assertNull(map.get("a"));
+        assertNull(map.remove("c"));
+    }
+
+    @Test
+    void testContainsKey() {
+        assertTrue(map.containsKey("a"));
+        assertFalse(map.containsKey("c"));
+        assertFalse(map.containsKey("123"));
+    }
+
+    @Test
+    void testContainsValue() {
+        assertTrue(map.containsValue("1"));
+        assertTrue(map.containsValue("2"));
+        assertFalse(map.containsValue("3"));
+        assertFalse(map.containsValue("123"));
+    }
+
+    @Test
+    void testPutAll() {
+        Map<String, String> bulk = Map.of("x", "10", "y", "20");
+        map.putAll(bulk);
+        assertEquals(4, map.size());
+        assertEquals("20", map.get("y"));
+    }
+
+    @Test
+    void testClear() {
+        map.clear();
+        assertTrue(map.isEmpty());
+    }
+
+    @Test
+    void testKeySet() {
+        assertEquals(Set.of("a", "b"), map.keySet());
+    }
+
+    @Test
+    void testValues() {
+        assertTrue(map.values().containsAll(List.of("1", "2")));
+    }
+
+    @Test
+    void testEntrySet() {
+        assertTrue(map.entrySet().contains(new AbstractMap.SimpleEntry<>("a", "1")));
+        assertEquals(2, map.entrySet().size());
+    }
+}