<div dir="ltr">I just took another look at the client - with the default (internal private) ExecutorService, it does an orderly shutdown() which won't interrupt the threads.<div><br></div><div>If you pass in your own ExecutorService to newConnection() you can call shutDownNow() which *will* interrupt the dispatcher threads (if they are doing something that's interruptible).</div>
<div><br></div><div>See ExecutorService.shutDownNow() for more information.</div><div><br></div><div>Gary</div></div><div class="gmail_extra"><br><br><div class="gmail_quote">On Wed, Apr 2, 2014 at 6:23 PM, Gary Russell <span dir="ltr"><<a href="mailto:grussell@gopivotal.com" target="_blank">grussell@gopivotal.com</a>></span> wrote:<br>
<blockquote class="gmail_quote" style="margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex"><div dir="ltr">In Spring-AMQP, we don't asynchronously cancel the consumer, we tell the consumer it needs to cancel itself when it's finished its current work. IIRC, if the consumer is running when you cancel it, its basicAck will fail because you've already cancelled it and the broker has re-queued that message. Effectively we tell the consumer to emit the cancel immediately after the ack.<div class="">
<div>
<br></div><div>><span style="font-family:arial,sans-serif;font-size:13px">stop a badly behaving Consumer</span></div><div><span style="font-family:arial,sans-serif;font-size:13px"><br></span></div></div><div><span style="font-family:arial,sans-serif;font-size:13px">It depends on how badly behaved it is; if it never executes any interruptible code (say it's stuck in a while() loop), you are out of luck. If it does stuff that's interruptible (acquire locks, etc, etc) then you can interrupt it.</span></div>
<div><span style="font-family:arial,sans-serif;font-size:13px"><br></span></div><div><span style="font-family:arial,sans-serif;font-size:13px">I don't know enough about the rabbit client internals as to whether there is a mechanism to force it to interrupt the dispatcher thread (I didn't see anything after a quick look) so you might have to roll your own (capture a reference to the thread in handleDelivery and interrupt it after your timeout).</span></div>
<div><br></div><div>When all else fails, System.exit() is your friend.</div><span class="HOEnZb"><font color="#888888"><div><br></div><div>Gary</div><div><br></div></font></span></div><div class="HOEnZb"><div class="h5">
<div class="gmail_extra"><br><br><div class="gmail_quote">On Wed, Apr 2, 2014 at 5:52 PM, Bertrand Guay-Paquet <span dir="ltr"><<a href="mailto:bernie@step.polymtl.ca" target="_blank">bernie@step.polymtl.ca</a>></span> wrote:<br>
<blockquote class="gmail_quote" style="margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex">Thanks Jason and Gary,<br>
<br>
Your replies really helped and I almost got it now. At first, I didn't know that handleCancelOk only runs when a consumer is not handling anything else. That is, only a single handle*() method can run at the same time on any given consumer instance. Thinking about it now, I realize this was explained in the doc with the phrase "Each Channel has its own dispatch thread."<br>
<br>
So I created my subclass of DefaultConsumer (BasicConsumer) which implements isCancelled() and handleCancelOk() which sets a "cancelled flag" to true. Is there already a built-in way to know if a consumer was cancelled? I couldn't find it myself...<br>
<br>
Here are my start and stop sequences which almost completely work. I haven't found how to forcefully stop a badly behaving Consumer after a given timeout at the end of the stop sequence. How can this be achieved?<br>
<br>
Start sequence:<br>
consumers = new ArrayList<BasicConsumer>();<br>
consumers.add(...)<br>
<br>
Stop sequence:<br>
// Cancel all consumers<br>
for (BasicConsumer consumer : consumers) {<br>
try {<br>
consumer.getChannel().<u></u>basicCancel(consumer.<u></u>getConsumerTag());<br>
} catch (Exception e) {<br>
// report<br>
}<br>
}<br>
<br>
// Wait for all consumers to be cancelled<br>
Timeout timeout = ...;<br>
while (!consumers.isEmpty() && !timeout.isElapsed()) {<br>
// Remove cancelled consumers<br>
for (Iterator<BasicConsumer> iterator = consumers.iterator(); iterator.hasNext();) {<br>
if (iterator.next().isCancelled()<u></u>)<br>
iterator.remove();<br>
}<br>
}<br>
<br>
// Force close remaining (timed-out) consumers<br>
for (BasicConsumer consumer : consumers) {<br>
consumer.getChannel().abort(); // <---- doesn'k kill the consumer<br>
// <---------------- How do I kill a timed-out consumer?<br>
}<br>
connection.close();<br>
<br>
Regards,<br>
Bertrand<br>
______________________________<u></u>_________________<br>
rabbitmq-discuss mailing list<br>
<a href="mailto:rabbitmq-discuss@lists.rabbitmq.com" target="_blank">rabbitmq-discuss@lists.<u></u>rabbitmq.com</a><br>
<a href="https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss" target="_blank">https://lists.rabbitmq.com/<u></u>cgi-bin/mailman/listinfo/<u></u>rabbitmq-discuss</a><br>
</blockquote></div><br></div>
</div></div></blockquote></div><br></div>