refactor: replace boolean parallelExecution with configurable numThreads

Address maintainer feedback on PR #191:

- Replace `boolean serialExecution` with `int numThreads` parameter
  in RespServerContext and RespServer.Builder
- StateHolder now accepts a Map implementation via constructor:
  HashMap for single-thread (numThreads=1), ConcurrentHashMap for
  multi-thread (numThreads>1)
- Scheduler always used (no bypass path), thread pool size matches
  numThreads: newSingleThreadExecutor for 1, newFixedThreadPool for >1
- Preserve upstream daemon thread naming ("resp-server")
- Remove processCommand() if/else branching — unified scheduler path

This eliminates the two concerns raised:
1. No unused scheduler in any mode
2. No unnecessary ConcurrentHashMap synchronization in single-thread mode

Made-with: Cursor

Author: haiyang.zhou

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

@@ -228,7 +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;
+    private int numThreads = 1;
 
     public Builder host(String host) {
       this.host = host;
@@ -256,19 +256,22 @@ public Builder commands(CommandSuite commands) {
     }
 
     /**
-     * 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.
+     * Sets the number of threads for the command execution scheduler.
+     * Default is 1 (single-thread, global serialization).
+     * Higher values enable parallel command execution with thread-safe state access.
+     *
+     * @param numThreads number of threads (must be >= 1)
      */
-    public Builder parallelExecution() {
-      this.serialExecution = false;
+    public Builder numThreads(int numThreads) {
+      if (numThreads < 1) {
+        throw new IllegalArgumentException("numThreads must be >= 1");
+      }
+      this.numThreads = numThreads;
       return this;
     }
 
     public RespServer build() {
-      return new RespServer(new RespServerContext(host, port, commands, SessionListener.nullListener(), serialExecution));
+      return new RespServer(new RespServerContext(host, port, commands, SessionListener.nullListener(), numThreads));
     }
   }
 }

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

@@ -8,8 +8,10 @@
 import static com.github.tonivade.resp.util.Precondition.checkNonNull;
 import static com.github.tonivade.resp.util.Precondition.checkRange;
 import static com.github.tonivade.resp.SessionListener.nullListener;
+import static java.util.concurrent.Executors.newFixedThreadPool;
 import static java.util.concurrent.Executors.newSingleThreadExecutor;
 
+import java.util.HashMap;
 import java.util.Optional;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.function.Function;
@@ -30,39 +32,37 @@
 public class RespServerContext implements ServerContext {
 
   private static final String RESP_SERVER = "resp-server";
+  private static final int DEFAULT_NUM_THREADS = 1;
 
   private static final Logger LOGGER = LoggerFactory.getLogger(RespServerContext.class);
 
-  private final StateHolder state = new StateHolder();
+  private final StateHolder state;
   private final ConcurrentHashMap<String, Session> clients = new ConcurrentHashMap<>();
-  private final Scheduler scheduler = Schedulers.from(newSingleThreadExecutor(runnable -> {
-    Thread thread = new Thread(runnable);
-    thread.setName(RESP_SERVER);
-    thread.setDaemon(true);
-    return thread;
-  }));
+  private final Scheduler scheduler;
 
   private final String host;
   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);
+    this(host, port, commands, sessionListener, DEFAULT_NUM_THREADS);
   }
 
   public RespServerContext(String host, int port, CommandSuite commands,
-      SessionListener sessionListener, boolean serialExecution) {
+      SessionListener sessionListener, int numThreads) {
     this.host = checkNonEmpty(host);
     this.port = checkRange(port, 1024, 65535);
     this.commands = checkNonNull(commands);
     this.sessionListener = checkNonNull(sessionListener);
-    this.serialExecution = serialExecution;
+    this.state = new StateHolder(numThreads > 1 ? new ConcurrentHashMap<>() : new HashMap<>());
+    this.scheduler = Schedulers.from(numThreads > 1
+        ? newFixedThreadPool(numThreads, runnable -> newDaemonThread(runnable, RESP_SERVER))
+        : newSingleThreadExecutor(runnable -> newDaemonThread(runnable, RESP_SERVER)));
   }
 
   public void start() {
@@ -121,21 +121,12 @@ void processCommand(Request request) {
     LOGGER.debug("received command: {}", request);
 
     var command = getCommand(request.getCommand());
-    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);
-      }
+    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);
     }
   }
 
@@ -173,4 +164,11 @@ private void clear() {
     clients.clear();
     state.clear();
   }
+
+  private static Thread newDaemonThread(Runnable runnable, String name) {
+    Thread thread = new Thread(runnable);
+    thread.setName(name);
+    thread.setDaemon(true);
+    return thread;
+  }
 }

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

@@ -4,13 +4,21 @@
  */
 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 ConcurrentHashMap<>();
+  private final Map<String, Object> state;
+
+  public StateHolder() {
+    this(new HashMap<>());
+  }
+
+  public StateHolder(Map<String, Object> state) {
+    this.state = state;
+  }
 
   @SuppressWarnings("unchecked")
   public <T> Optional<T> getValue(String key) {

src/test/java/com/github/tonivade/resp/RespServerContextTest.java

@@ -160,31 +160,31 @@ public void requireCallback() {
   }
 
   @Test
-  public void processCommandParallelMode() {
-    var parallelContext = new RespServerContext(HOST, PORT, commands,
-        SessionListener.nullListener(), false);
-    Request request = new DefaultRequest(parallelContext, session, safeString("test"), Collections.emptyList());
+  public void processCommandMultiThreaded() {
+    var multiThreadContext = new RespServerContext(HOST, PORT, commands,
+        SessionListener.nullListener(), 4);
+    Request request = new DefaultRequest(multiThreadContext, session, safeString("test"), Collections.emptyList());
     when(commands.getCommand(request.getCommand())).thenReturn(respCommand);
     when(respCommand.execute(request)).thenReturn(nullString());
 
-    parallelContext.processCommand(request);
+    multiThreadContext.processCommand(request);
 
-    verify(respCommand).execute(request);
-    verify(session).publish(nullString());
+    verify(respCommand, timeout(1000)).execute(request);
+    verify(session, timeout(1000)).publish(nullString());
   }
 
   @Test
-  public void processCommandParallelModeException() {
-    var parallelContext = new RespServerContext(HOST, PORT, commands,
-        SessionListener.nullListener(), false);
-    Request request = new DefaultRequest(parallelContext, session, safeString("test"), Collections.emptyList());
+  public void processCommandMultiThreadedException() {
+    var multiThreadContext = new RespServerContext(HOST, PORT, commands,
+        SessionListener.nullListener(), 4);
+    Request request = new DefaultRequest(multiThreadContext, session, safeString("test"), Collections.emptyList());
     when(commands.getCommand(request.getCommand())).thenReturn(respCommand);
     doThrow(RuntimeException.class).when(respCommand).execute(request);
 
-    parallelContext.processCommand(request);
+    multiThreadContext.processCommand(request);
 
-    verify(respCommand).execute(request);
-    verify(session, times(0)).publish(any());
+    verify(respCommand, timeout(1000)).execute(request);
+    verify(session, timeout(1000).atLeast(0)).publish(any());
   }
 
   private Request newRequest(String command) {