Close alibaba#75

添加独立线程解决update和commit/rollback互相争夺线程池资源以及数据库锁资源问题

Author: zhuchao941

server/src/main/config/com/alibaba/cobar/config/model/SystemConfig.java

@@ -46,6 +46,7 @@ public final class SystemConfig {
     private int processors;
     private int processorHandler;
     private int processorExecutor;
+    private int processorCommitter;
     private int initExecutor;
     private int timerExecutor;
     private int managerExecutor;
@@ -69,6 +70,7 @@ public SystemConfig() {
         this.processors = DEFAULT_PROCESSORS;
         this.processorHandler = DEFAULT_PROCESSORS;
         this.processorExecutor = DEFAULT_PROCESSORS;
+        this.processorCommitter = DEFAULT_PROCESSORS;
         this.managerExecutor = DEFAULT_PROCESSORS;
         this.timerExecutor = DEFAULT_PROCESSORS;
         this.initExecutor = DEFAULT_PROCESSORS;
@@ -133,8 +135,16 @@ public int getProcessorExecutor() {
     public void setProcessorExecutor(int processorExecutor) {
         this.processorExecutor = processorExecutor;
     }
+    
+	public int getProcessorCommitter() {
+		return processorCommitter;
+	}
 
-    public int getManagerExecutor() {
+	public void setProcessorCommitter(int processorCommitter) {
+		this.processorCommitter = processorCommitter;
+	}
+
+	public int getManagerExecutor() {
         return managerExecutor;
     }
 

server/src/main/net/com/alibaba/cobar/net/NIOProcessor.java

@@ -39,26 +39,28 @@ public final class NIOProcessor {
     private final BufferPool bufferPool;
     private final NameableExecutor handler;
     private final NameableExecutor executor;
+    private final NameableExecutor committer;
     private final ConcurrentMap<Long, FrontendConnection> frontends;
     private final ConcurrentMap<Long, BackendConnection> backends;
     private final CommandCount commands;
     private long netInBytes;
     private long netOutBytes;
 
     public NIOProcessor(String name) throws IOException {
-        this(name, DEFAULT_BUFFER_SIZE, DEFAULT_BUFFER_CHUNK_SIZE, AVAILABLE_PROCESSORS, AVAILABLE_PROCESSORS);
+        this(name, DEFAULT_BUFFER_SIZE, DEFAULT_BUFFER_CHUNK_SIZE, AVAILABLE_PROCESSORS, AVAILABLE_PROCESSORS, AVAILABLE_PROCESSORS);
     }
 
-    public NIOProcessor(String name, int handler, int executor) throws IOException {
-        this(name, DEFAULT_BUFFER_SIZE, DEFAULT_BUFFER_CHUNK_SIZE, handler, executor);
+    public NIOProcessor(String name, int handler, int executor, int committer) throws IOException {
+        this(name, DEFAULT_BUFFER_SIZE, DEFAULT_BUFFER_CHUNK_SIZE, handler, executor, committer);
     }
 
-    public NIOProcessor(String name, int buffer, int chunk, int handler, int executor) throws IOException {
+    public NIOProcessor(String name, int buffer, int chunk, int handler, int executor, int committer) throws IOException {
         this.name = name;
         this.reactor = new NIOReactor(name);
         this.bufferPool = new BufferPool(buffer, chunk);
         this.handler = (handler > 0) ? ExecutorUtil.create(name + "-H", handler) : null;
         this.executor = (executor > 0) ? ExecutorUtil.create(name + "-E", executor) : null;
+        this.committer = (committer > 0) ? ExecutorUtil.create(name + "-C", committer) : null;
         this.frontends = new ConcurrentHashMap<Long, FrontendConnection>();
         this.backends = new ConcurrentHashMap<Long, BackendConnection>();
         this.commands = new CommandCount();
@@ -87,8 +89,12 @@ public NameableExecutor getHandler() {
     public NameableExecutor getExecutor() {
         return executor;
     }
+    
+	public NameableExecutor getCommitter() {
+		return committer;
+	}
 
-    public void startup() {
+	public void startup() {
         reactor.startup();
     }
 

server/src/main/server/com/alibaba/cobar/CobarServer.java

@@ -104,9 +104,10 @@ public void startup() throws IOException {
         LOGGER.info("Startup processors ...");
         int handler = system.getProcessorHandler();
         int executor = system.getProcessorExecutor();
+        int committer = system.getProcessorCommitter();
         processors = new NIOProcessor[system.getProcessors()];
         for (int i = 0; i < processors.length; i++) {
-            processors[i] = new NIOProcessor("Processor" + i, handler, executor);
+            processors[i] = new NIOProcessor("Processor" + i, handler, executor, committer);
             processors[i].startup();
         }
         timer.schedule(processorCheck(), 0L, system.getProcessorCheckPeriod());

server/src/main/server/com/alibaba/cobar/manager/response/ShowThreadPool.java

@@ -125,6 +125,7 @@ private static List<NameableExecutor> getExecutors() {
         for (NIOProcessor p : server.getProcessors()) {
             list.add(p.getHandler());
             list.add(p.getExecutor());
+            list.add(p.getCommitter());
         }
         return list;
     }

server/src/main/server/com/alibaba/cobar/mysql/bio/executor/DefaultCommitExecutor.java

@@ -121,7 +121,7 @@ public void commit(final OkPacket packet, final BlockingSession session, final i
 
         // 执行
         final ConcurrentMap<RouteResultsetNode, Channel> target = session.getTarget();
-        Executor executor = session.getSource().getProcessor().getExecutor();
+        Executor committer = session.getSource().getProcessor().getCommitter();
         int started = 0;
         for (RouteResultsetNode rrn : target.keySet()) {
             if (rrn == null) {
@@ -136,7 +136,7 @@ public void commit(final OkPacket packet, final BlockingSession session, final i
             final MySQLChannel mc = (MySQLChannel) target.get(rrn);
             if (mc != null) {
                 mc.setRunning(true);
-                executor.execute(new Runnable() {
+                committer.execute(new Runnable() {
                     @Override
                     public void run() {
                         _commit(mc, session);

server/src/main/server/com/alibaba/cobar/mysql/bio/executor/RollbackExecutor.java

@@ -118,14 +118,14 @@ public void rollback(final BlockingSession session) {
         }
 
         // 执行
-        Executor exec = source.getProcessor().getExecutor();
+        Executor committer = source.getProcessor().getCommitter();
 
         int started = 0;
         for (RouteResultsetNode rrn : target.keySet()) {
             final MySQLChannel mc = (MySQLChannel) target.get(rrn);
             if (mc != null) {
                 mc.setRunning(true);
-                exec.execute(new Runnable() {
+                committer.execute(new Runnable() {
                     @Override
                     public void run() {
                         _rollback(mc, session);

server/src/test/resources/server.xml

@@ -28,6 +28,7 @@
     <property name="processors">4</property>
     <property name="processorHandler">8</property>
     <property name="processorExecutor">8</property>
+    <property name="processorCommitter">8</property>
     <property name="clusterHeartbeatUser">_HEARTBEAT_USER_</property>
     <property name="clusterHeartbeatPass">_HEARTBEAT_PASS_</property>
   </system>