Using RabbitMQ with Plone - Celery or not?

celery, plone, python, rabbitmq

Solution

At first, ask yourself, if you need the features of RabbitMQ or just want to do some asynchronous tasks in Python with Plone.

If you don't really need RabbitMQ, you could look into David Glick's gists for how to integrate Celery with Plone (and still use RabbitMQ with Celery):

- https://gist.github.com/davisagli/5824662

- https://gist.github.com/davisagli/5824709

You could also look into collective.taskqueue (simple queues without Celery nor RabbitMQ), but it does not provide any monitoring solution yet.

If you really need RabbitMQ, skip Celery, and try out collective.zamqp. Celery tries to be broker by itself and would prevent you from using most of AMQP's and RabbitMQ's built-in features.

RabbitMQ ships with great web admin plugin for monitoring and there are also plugins for 3rd party monitoring systems (like Zenoss).

I'm sorry that collective.zamqp is still missing narrative documentation, but you can look into collective.zamqpdemo for various examples of its configuration and usage.

In short, c.zamqp allows you to define configure broker usage in terms of producers and consumers:

from five import grok
from zope.interface import Interface
from collective.zamqp.producer import Producer
from collective.zamqp.consumer import Consumer


class CreateItemProducer(Producer):
    """Produces item creation requests"""
    grok.name("amqpdemo.create")  # is also used as default routing key

    connection_id = "superuser"
    serializer = "msgpack"
    queue = "amqpdemo.create"

    durable = False


class ICreateItemMessage(Interface):
    """Marker interface for item creation message"""


class CreateItemConsumer(Consumer):
    """Consumes item creation messages"""
    grok.name("amqpdemo.create")  # is also used as the queue name

    connection_id = "superuser"
    marker = ICreateItemMessage

    durable = False

Publish messages through transaction bound producer (to publish messages only after a successful transaction):

    import uuid

    from zope.component import getUtility
    from collective.zamqp.interfaces import IProducer

    producer = getUtility(IProducer, name="amqpdemo.create")
    producer._register()  # register to bound to successful transaction

    message = {"title": u"My title"}

    producer.publish(message)

And consume the messages in a familiar content event handler environment:

from zope.component.hooks import getSite
from collective.zamqp.interfaces import IMessageArrivedEvent
from plone.dexterity.utils import createContentInContainer

@grok.subscribe(ICreateItemMessage, IMessageArrivedEvent)
def createItem(message, event):
    """Consume item creation message"""

    portal = getSite()
    obj = createContentInContainer(
        portal, "Document", checkConstraints=True, **message.body)

    message.ack()

Finally, it decouples broker connection configuration from code and the actual connection parameters can be defined in buildout.cfg (allowing required amount of consuming instances):

[instance]
recipe = plone.recipe.zope2instance
...
zope-conf-additional =
    %import collective.zamqp
    <amqp-broker-connection>
         connection_id superuser
         heartbeat 120
# These are defaults, but can be defined when required:     
#        hostname localhost
#        virtual_host /
#        username guest
#        password guest
    </amqp-broker-connection>
    <amqp-consuming-server>
        connection_id superuser
        site_id Plone
        user_id admin
        vhm_method_prefix /VirtualHostBase/https/example.com:443/Plone/VirtualHostRoot
    </amqp-consuming-server>

c.zamqp cannot be directly called from RestrictedPython, so integrating it to PloneFormGen would need either a custom action adapter or a custom External method to be called from PFG's Python script adapter.

Problem

Original source