feat: add configurable parallel execution mode

Add a parallelExecution() option to RespServer.Builder that controls
the command execution threading model:

- Serial mode (default): commands are enqueued to a single-thread
  RxJava scheduler, providing global serialization. This preserves
  backward compatibility.

- Parallel mode: commands execute directly on Netty I/O threads,
  giving per-connection serialization via EventLoop binding. This
  yields higher throughput for stateless or thread-safe command
  implementations.

StateHolder is changed from HashMap to ConcurrentHashMap to ensure
correctness when parallel execution is enabled and multiple I/O
threads may access shared state concurrently.

Made-with: Cursor

Author: haiyang.zhou

src/main/java/com/github/tonivade/resp/RespServer.java

@@ -228,6 +228,7 @@ public static class Builder implements Recoverable {
     private String host = DEFAULT_HOST;
     private int port = DEFAULT_PORT;
     private CommandSuite commands = new CommandSuite();
+    private boolean serialExecution = true;
 
     public Builder host(String host) {
       this.host = host;
@@ -254,8 +255,20 @@ public Builder commands(CommandSuite commands) {
       return this;
     }
 
+    /**
+     * When enabled (default), all commands are executed on a single-thread
+     * executor, providing global serialization. When disabled, commands
+     * execute directly on Netty I/O threads, giving per-connection
+     * serialization with higher throughput for stateless or thread-safe
+     * command implementations.
+     */
+    public Builder parallelExecution() {
+      this.serialExecution = false;
+      return this;
+    }
+
     public RespServer build() {
-      return new RespServer(new RespServerContext(host, port, commands));
+      return new RespServer(new RespServerContext(host, port, commands, SessionListener.nullListener(), serialExecution));
     }
   }
 }

src/main/java/com/github/tonivade/resp/RespServerContext.java

@@ -38,16 +38,23 @@ public class RespServerContext implements ServerContext {
   private final int port;
   private final CommandSuite commands;
   private final SessionListener sessionListener;
+  private final boolean serialExecution;
 
   public RespServerContext(String host, int port, CommandSuite commands) {
     this(host, port, commands, nullListener());
   }
 
   public RespServerContext(String host, int port, CommandSuite commands, SessionListener sessionListener) {
+    this(host, port, commands, sessionListener, true);
+  }
+
+  public RespServerContext(String host, int port, CommandSuite commands,
+      SessionListener sessionListener, boolean serialExecution) {
     this.host = checkNonEmpty(host);
     this.port = checkRange(port, 1024, 65535);
     this.commands = checkNonNull(commands);
     this.sessionListener = checkNonNull(sessionListener);
+    this.serialExecution = serialExecution;
   }
 
   public void start() {
@@ -106,12 +113,21 @@ void processCommand(Request request) {
     LOGGER.debug("received command: {}", request);
 
     var command = getCommand(request.getCommand());
-    try {
-      enqueue(Observable.fromCallable(() -> executeCommand(command, request)))
-        .subscribe(response -> processResponse(request, response),
-                   ex -> LOGGER.error("error executing command: " + request, ex));
-    } catch (RuntimeException ex) {
-      LOGGER.error("error executing command: " + request, ex);
+    if (serialExecution) {
+      try {
+        enqueue(Observable.fromCallable(() -> executeCommand(command, request)))
+          .subscribe(response -> processResponse(request, response),
+                     ex -> LOGGER.error("error executing command: " + request, ex));
+      } catch (RuntimeException ex) {
+        LOGGER.error("error executing command: " + request, ex);
+      }
+    } else {
+      try {
+        RedisToken response = executeCommand(command, request);
+        processResponse(request, response);
+      } catch (RuntimeException ex) {
+        LOGGER.error("error executing command: " + request, ex);
+      }
     }
   }
 

src/main/java/com/github/tonivade/resp/StateHolder.java

@@ -4,13 +4,13 @@
  */
 package com.github.tonivade.resp;
 
-import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
 
 public class StateHolder {
 
-  private final Map<String, Object> state = new HashMap<>();
+  private final Map<String, Object> state = new ConcurrentHashMap<>();
 
   @SuppressWarnings("unchecked")
   public <T> Optional<T> getValue(String key) {