This is related to QPID-1479.

For starters I have changed the IoSender.java IoReceiver.java and AMQSession.java#Dispatcher to use the Thread factory to create the threads they require.
The ThreadFactory has two implimentations, the default being the java.lang.Threads.
The other is the RealtimeThreadFactory which uses reflection to create threads with a specific priority.
-Dqpid.thread_factory=<thread_factory_class> will decide which thread factory should be loaded.
-Dqpid.rt_thread_priority=<int> specifies the gloabl real time thread priority and defaults to 20.
You could also set individual thread priorities by adding the nessacery config+code changes.

I have also changed the Testkit and QpidBench to use the Thread factory so you could use them for testing/benchmarking work on RT JVMs.



git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@719628 13f79535-47bb-0310-9956-ffa450edef68

1 files changed, 54 insertions, 11 deletions

diff --git a/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java b/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java
index 7411e81bd6..4bba7b113d 100644
--- a/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java
+++ b/java/tools/src/main/java/org/apache/qpid/tools/QpidBench.java
@@ -20,23 +20,45 @@
  */
 package org.apache.qpid.tools;
 
-import java.lang.reflect.InvocationTargetException;
+import static org.apache.qpid.tools.QpidBench.Mode.BOTH;
+import static org.apache.qpid.tools.QpidBench.Mode.CONSUME;
+import static org.apache.qpid.tools.QpidBench.Mode.PUBLISH;
+
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.UUID;
-import javax.jms.*;
+
+import javax.jms.DeliveryMode;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.TextMessage;
 
 import org.apache.qpid.client.AMQConnection;
-import org.apache.qpid.transport.*;
-import org.apache.qpid.transport.network.io.IoTransport;
+import org.apache.qpid.thread.Threading;
+import org.apache.qpid.transport.DeliveryProperties;
+import org.apache.qpid.transport.ExchangeBind;
+import org.apache.qpid.transport.Header;
+import org.apache.qpid.transport.MessageAcceptMode;
+import org.apache.qpid.transport.MessageAcquireMode;
+import org.apache.qpid.transport.MessageCreditUnit;
+import org.apache.qpid.transport.MessageDeliveryMode;
+import org.apache.qpid.transport.MessageFlowMode;
+import org.apache.qpid.transport.MessageProperties;
+import org.apache.qpid.transport.MessageSubscribe;
+import org.apache.qpid.transport.MessageTransfer;
+import org.apache.qpid.transport.QueueDeclare;
+import org.apache.qpid.transport.SessionException;
+import org.apache.qpid.transport.SessionListener;
 import org.apache.qpid.util.UUIDGen;
 import org.apache.qpid.util.UUIDs;
 
-import static org.apache.qpid.tools.QpidBench.Mode.*;
-
 /**
  * QpidBench
  *
@@ -412,7 +434,7 @@ public class QpidBench
         {
         case CONSUME:
         case BOTH:
-            new Thread()
+            Runnable r = new Runnable()
             {
                 public void run()
                 {
@@ -432,7 +454,18 @@ public class QpidBench
                         throw new RuntimeException(e);
                     }
                 }
-            }.start();
+            };
+           
+            Thread t;
+            try
+            {
+                t = Threading.getThreadFactory().createThread(r);                      
+            }
+            catch(Exception e)
+            {
+                throw new Error("Error creating consumer thread",e);
+            }
+            t.start();
             break;
         }
 
@@ -440,7 +473,7 @@ public class QpidBench
         {
         case PUBLISH:
         case BOTH:
-            new Thread()
+            Runnable r = new Runnable()
             {
                 public void run()
                 {
@@ -460,7 +493,17 @@ public class QpidBench
                         throw new RuntimeException(e);
                     }
                 }
-            }.start();
+            };
+            Thread t;
+            try
+            {
+                t = Threading.getThreadFactory().createThread(r);                      
+            }
+            catch(Exception e)
+            {
+                throw new Error("Error creating publisher thread",e);
+            }
+            t.start();
             break;
         }
     }