How to detect an emergency blocking RabbitMQ manufacturer?

Question

I have a producer who sends long-term messages to the RabbitMQ exchange. If the memory or disk of RabbitMQ exceeds the watermark threshold, RabbitMQ will block my manufacturer. The documentation says that it stops reading data from the socket, and also stops the pulse.

I would like to know that I am blocked in my manufacturer code. Nowadays, even with the pulse on, everything just stops forever. I would like to get some kind of exception, so that I know that I am blocked, and I can warn the user and / or take some other action, but I can not find a way to do this. I use both Java and C # clients and need this feature in both. Any advice? Thanks.

Answer 1

I apologize, but with RabbitMQ (at least from 2.8.6) this is not possible: - (

had a similar problem that centered around trying to establish a channel when the connection was blocked. The result was the same as what you are experiencing.

I did some research on the actual core of the RabbitMQ C # .Net library and found the root cause of the problem is that it goes into an infinite lock state.

You can see more information on the RabbitMQ mailing list here:

http://rabbitmq.1065348.n5.nabble.com/Net-Client-locks-trying-to-create-a-channel-on-a-blocked-connection-td21588.html

One suggestion (which we did not implement) was to do the work inside the thread and have some other component that controls the timeout and kills the thread if it is exceeded. We just took the risk: - (

Answer 2

Rabbitmq uses an rpc blocking call that waits for an indefinite response.

If you look at the Java client api, then what it does:

AMQChannel.BlockingRpcContinuation k = new AMQChannel.SimpleBlockingRpcContinuation(); k.getReply(-1); 

Now -1 is passed in blocks of arguments until a response is received.

It's good that you can go to your timeout to return it. The bad thing is that you have to update client banks.

If this is all right, you can time out wherever the blocking call is made, as indicated above. The code will look something like this:

 try { return k.getReply(200); } catch (TimeoutException e) { throw new MyCustomRuntimeorTimeoutException("RabbitTimeout ex",e); } 

And in your code, you can handle this exception and execute your logic in this case.

Some related classes that may require this fix will be as follows:

 com.rabbitmq.client.impl.AMQChannel com.rabbitmq.client.impl.ChannelN com.rabbitmq.client.impl.AMQConnection 

FYI: I tried this and it works.