CODEWITHSUNDEEP
callbackrabbitmqpublish-subscribeconsumerrabbitmq-exchange
javaguy
javaguy

Reputation: 31

RabbitMQ publish to multiple queues via an exchange publishes to only the first queue only

I want to publish to Several queues bound to an exchange (fan out will not work here as I want to publish to only selective queues and not all the queues)

The code goes like this:

public int pushDataOverRabbitMQ(String[] deviceIDs, String msg) throws IOException {
        int retrunVal=0;
        String EXCHANGE_NAME="DUSHTEST";
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();


        channel.exchangeDeclare(EXCHANGE_NAME, "direct",true);

        for(int i=0;i<deviceIDs.length;i++){

            try{
                channel.queueDeclare(deviceIDs[i],true,false,false,null).getQueue();


                channel.queueBind(deviceIDs[i], EXCHANGE_NAME, deviceIDs[i]);
//routing key and queue have the same value

                channel.basicPublish(EXCHANGE_NAME, deviceIDs[i],  new AMQP.BasicProperties.Builder()
                .contentType("text/plain").deliveryMode(1)
                .priority(1).userId("guest")
                .build(),
                msg.getBytes());

                System.out.println(" [x] Sent '" + msg+ "'");

                retrunVal= Constants.PUB_STATUS_SUCCESS;
            }catch(Exception e){
                e.printStackTrace();
                retrunVal= Constants.PUB_STATUS_FAIL;
            }
        }
        channel.close();
        connection.close();
        return retrunVal;

    }

**deviceIDs** contains two Strings Dush-Micromax and MyDesktop

I am unable to upload images due to low points, but on the server side there are three queues created. The code above binds the created queues to the exchange.

I had earlier created two consumers as below:

import java.io.IOException;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;


public class RabbitSubscriptionModel {

    /**
     * @param args
     */
    private final static String QUEUE_NAME = "MyDesktop";
    //  private final static String QUEUE_NAME = "Dush-Micromax";
    public static void main(String[] args) throws IOException {
        // TODO Auto-generated method stub
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();
        try{
            channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        }catch (Exception e) {
            e.printStackTrace();
        }


        channel.basicConsume(QUEUE_NAME, false,"My-Consumer-Trial1", new GenericConsumer(channel));
//      channel.basicConsume(QUEUE_NAME, false,"My-Consumer-Trial2", new GenericConsumer(channel));

    }

}

I have two Consumers namely My-Consumer-Trial1 and My-Consumer-Trial2

However, I receive data only in the first consumer ie: My-Consumer-Trial1. After carefully analyzing the rabbitMQ Admin Console it seems that instead of publishing to the queue that the consumers is hooked onto the second publish happens on a new Queue. So in total we have three queues of which two are bound to the EXCHANGE and one under default EXCHANGE.

Upvotes: 0

Views: 4542

Answers (1)

robthewolf
robthewolf

Reputation: 7624

You are missing a lot here in your code. For reference check out this tutorial on direct exchanges http://www.rabbitmq.com/tutorials/tutorial-four-java.html.

You are not connecting to the exchange in the consumer

You are doing queueDeclare queueBind in the producer and not the consumer.

You have no while loop to read the messages

Upvotes: 1

Related Questions