Hi Everyone,<div><br></div><div>I'm playing around with Rabbitmq(sorry for flooding the list lately) and am having some weird problems in my tests. I have a multi-threaded application that I was considering using rabbitmq to manage the queue but I'm getting alot of blocking between my channels. Is there a preferred way to setup pools?
</div><div><br></div><div>I'm a bit new to Java so I used to the pools from apache commons but when I profile the channels are all blocking each other. To test if dedicated connections per thread worked better I made an example of it as well and it does not block when the same program has a single channel per connection. </div>
<div><br></div><div>So my question, do channels block each other or am I doing something wrong(using java api wrong or misunderstanding connection/channels)? Is there a more preferred way? Unrelated to the question, but I was also wondering is there any difference between channels and connections in terms of throughput(i.e. would there be any benefit of using dedicated connections ignoring the overhead in establishing/maintaining the connection)?</div>
<div><br></div><div>Here's the test code(it simply sends to a thread a bunch of data to write the queue). If it helps I'm using v2.7.1, installed from MacPorts):</div><div><br></div><div><br></div><div><div>import org.apache.commons.pool.BasePoolableObjectFactory;</div>
<div>import org.apache.commons.pool.ObjectPool;</div><div>import org.apache.commons.pool.PoolableObjectFactory;</div><div>import org.apache.commons.pool.impl.GenericObjectPool;</div><div><br></div><div>import java.io.IOException;</div>
<div>import java.util.NoSuchElementException;</div><div>import java.util.Set;</div><div>import java.util.concurrent.ExecutorService;</div><div>import java.util.concurrent.Executors;</div><div>import java.util.concurrent.LinkedBlockingDeque;</div>
<div>import java.util.concurrent.ThreadPoolExecutor;</div><div>import java.util.concurrent.TimeUnit;</div><div><br></div><div><br></div><div>import com.rabbitmq.client.ConnectionFactory;</div><div>import com.rabbitmq.client.Connection;</div>
<div>import com.rabbitmq.client.Channel;</div><div>import com.rabbitmq.client.MessageProperties;</div><div><br></div><div>public class PoolExample {</div><div> </div><div><span class="Apple-tab-span" style="white-space:pre"> </span>private static ExecutorService executor_worker;</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>static {</div><div> final int numberOfThreads_ThreadPoolExecutor = 20;</div>
<div> executor_worker = Executors.newFixedThreadPool(numberOfThreads_ThreadPoolExecutor);</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div>
<div> public static void main(String[] args) throws Exception {</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>System.out.println("starting.."); <span class="Apple-tab-span" style="white-space:pre"> </span> <span class="Apple-tab-span" style="white-space:pre"> </span></div>
<div> <span class="Apple-tab-span" style="white-space:pre"> </span>ObjectPool<Channel> pool =</div><div> new GenericObjectPool<Channel>(</div><div> new ConnectionPoolableObjectFactory(), 50);</div>
<div> <span class="Apple-tab-span" style="white-space:pre"> </span>for (int x = 0; x<500000000; x++) {</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>executor_worker.submit(new MyRunnable(x, pool));</div>
<div> <span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div> }</div><div>}</div><div><br></div><div>/*</div><div> //this is a version that creates its own connection per channel</div><div> class ConnectionPoolableObjectFactory</div>
<div> extends BasePoolableObjectFactory<Channel> {</div><div><br></div><div> private static final ConnectionFactory factory = newConnectionFactory();</div><div><br></div><div> private static ConnectionFactory newConnectionFactory() {</div>
<div> ConnectionFactory factory = new ConnectionFactory();</div><div> factory.setHost("localhost");</div><div> return factory;</div><div> }</div><div><br></div><div> @Override</div><div>
public Channel makeObject() throws Exception {</div><div> System.out.println("new channel");</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>return factory.newConnection().createChannel();</div>
<div> </div><div> }</div><div><br></div><div> @Override</div><div> public boolean validateObject(Channel channel) {</div><div> return channel.isOpen();</div><div> }</div><div><br></div><div> @Override</div>
<div> public void destroyObject(Channel channel) throws Exception {</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>System.out.println("closing");</div><div> channel.close();</div>
<div> }</div><div><br></div><div> @Override</div><div> public void passivateObject(Channel channel) throws Exception {</div><div> //System.out.println("sent back to queue");</div><div> }</div>
<div>}</div><div>*/</div><div><br></div><div>//this version pools channels </div><div>class ConnectionPoolableObjectFactory extends BasePoolableObjectFactory<Channel> {</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>private final Connection connection;</div>
<div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>ConnectionPoolableObjectFactory() throws IOException {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>ConnectionFactory factory = new ConnectionFactory();</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>factory.setHost("localhost");</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>connection = factory.newConnection();</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>@Override</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>public Channel makeObject() throws Exception {</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>return connection.createChannel();</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>@Override</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>public boolean validateObject(Channel channel) {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>return channel.isOpen();</div><div>
<span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>@Override</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>public void destroyObject(Channel channel) throws Exception {</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>channel.close();</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>@Override</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>public void passivateObject(Channel channel) throws Exception {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>//do nothing</div><div>
<span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div>}</div><div><br></div><div><br></div><div>class MyRunnable implements Runnable{ </div><div><span class="Apple-tab-span" style="white-space:pre"> </span>protected int x = 0;</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>protected ObjectPool<Channel> pool;</div><div><span class="Apple-tab-span" style="white-space:pre"> </span></div><div> public MyRunnable(int x, ObjectPool<Channel> pool) {</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>// TODO Auto-generated constructor stub</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>this.x = x;</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>this.pool = pool;</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>}</div><div><br></div><div><span class="Apple-tab-span" style="white-space:pre"> </span>public void run(){</div><div> try {</div><div> Channel channel = pool.borrowObject();</div>
<div> <span class="Apple-tab-span" style="white-space:pre"> </span>String message = Integer.toString(x);</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>channel.basicPublish( "", "task_queue", </div>
<div> MessageProperties.PERSISTENT_TEXT_PLAIN,</div><div> message.getBytes());</div><div> <span class="Apple-tab-span" style="white-space:pre"> </span>pool.returnObject(channel);</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>} catch (NoSuchElementException e) {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>// TODO Auto-generated catch block</div><div>
<span class="Apple-tab-span" style="white-space:pre"> </span>e.printStackTrace();</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>} catch (IllegalStateException e) {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>// TODO Auto-generated catch block</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>e.printStackTrace();</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>} catch (Exception e) {</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>// TODO Auto-generated catch block</div>
<div><span class="Apple-tab-span" style="white-space:pre"> </span>e.printStackTrace();</div><div><span class="Apple-tab-span" style="white-space:pre"> </span>} </div><div> }</div><div>}</div><div><br></div></div>