Publish multiple messages to RabbitMQ from a file

Question

Publishing single messages to a RabbitMQ queue can be easily done with the UI, by simply putting the message in the UI and clicking the "Publish Message" button.

How do you publish a batch of messages?

I have a file with messages to be sent to RabbitMQ. Each line has one message.

How can I publish all the messages from the file to my RabbitMQ server?

Is there a way to do it from command line?

Solution 1

Using rabbitmqadmin

while read -r line; do 
  echo $line | rabbitmqadmin publish exchange=amq.default routing_key=my_queue ; 
done < messages

Not specifying the payload parameter to rabbitmqadmin publish means it reads the payload from stdin.

Solution 2

you can use curl and rabbitmq api:

curl -u login:pass -i -H "content-type:application/json" -X POST http://localhost:15672/api/exchanges/%2Fvhost/exchange/publish -d'{"properties":{},"routing_key":"","payload":"you message","payload_encoding":"string"}'

this is an example of one message, by analogy, you can write a script

Solution 3

You need to use rabbitmqadmin cli tool:

https://www.rabbitmq.com/management-cli.html

rabbitmqadmin publish exchange=amq.default routing_key=test payload="hello, world"

Solution 4

As a variation to the answer from looseend you could also use GNU Parallel

This will yield far better performance if you have a large file.

  cat messages | parallel -j 100 \
  ./rabbitmqadmin -H $RABBITMQ_HOST \
                  -u $RABBITMQ_USERNAME \
                  -p $RABBITMQ_PASSWORD  \
                  publish exchange=amq.default \
                  routing_key=myqueue \
                  payload="{}"

This will run with 100 jobs. Omit host and credentials if not needed.

Solution 5

I've updated rabbitmqadmin file to support file content publishing. Try to find line containing EXTRA_VERBS = { as well as def invoke_publish(self): and update their related code as follows

EXTRA_VERBS = {
    'publish': {'mandatory': ['routing_key'],
                'optional':  {'payload': None,
                              'pfile': None,
                              'properties': {},
                              'exchange': 'amq.default',
                              'payload_encoding': 'string'},
                'json':      ['properties'],
                'uri':       '/exchanges/{vhost}/{exchange}/publish'},
    'get':     {'mandatory': ['queue'],
                'optional':  {'count': '1', 'requeue': 'true',
                              'payload_file': None, 'encoding': 'auto'},
                'uri':       '/queues/{vhost}/{queue}/get'}
}

and

def invoke_publish(self):
    (uri, upload) = self.parse_args(self.args, EXTRA_VERBS['publish'])
    if not 'payload' and 'pfile' in upload:
        data = sys.stdin.read()
        upload['payload'] = b64(data)
        upload['payload_encoding'] = 'base64'
    elif not 'payload' in upload:
        with open('populate/' + upload['pfile']) as f: data = f.read()
        upload['payload'] = b64(data)
        upload['payload_encoding'] = 'base64'
    resp = json.loads(self.post(uri, json.dumps(upload)))
    if resp['routed']:
        self.verbose("Message published")
    else:
        self.verbose("Message published but NOT routed")

Remove 'populate/' + from the following line if you want to provide file using absolute path.

with open('populate/' + upload['pfile']) as f: data = f.read()

Without updates to open(...) following command worked for me fine, assuming that file rules.json was placed in relative directory "populate"

python rabbitmqadmin.py publish exchange=feed-mgmt-in routing_key='#' properties='{"type":"domain-collections/rules"}' pfile="rules.json"

View more solutions

Comments

[summerbulb]

How does this snippet publish the messages/payloads from the file?

[aeryaguzov]

try smth like this: while read line; do rabbitmqadmin publish exchange=amq.default routing_key=test payload="$line"; done < file

[Boris Berkgaut]

payload="$line" would break JSON \-escapes like {"foo": "\"nested quotes\""} (because you shell also interprets \-escapes). But you can pass payload to rabbitmqadmin's stdin.

[joekarl]

Yeah, as noted it's unfortunate that rabbitmqadmin requires input to be text (due to input from the shell), not to mention that it uses the http api for publishing/consuming so it tends to be pretty slow for larger usage.

[glmrenard]

I tried but get (node:24786) MaxListenersExceededWarning: Possible EventEmitter memory leak detected. 11 error listeners added. Use emitter.setMaxListeners() to increase limit

[glmrenard]

With this it went 5 time quicker :)

[Hus Mukh]

I also tried but its giving error for routingKey even routing key is provided.