Combineer meerdere programmeertalen met RabbitMQ

Veel teams kiezen een programmeertaal die past bij het project waar ze aan werken. De argumenten voor het kiezen van een programmeertaal kunnen ver uiteenlopen, van leesbaarheid tot performance en security. Bij specifieke projecten kunnen bepaalde libraries nodig zijn, die niet beschikbaar zijn voor alle programmeertalen. Dit is ook het geval bij het gebruik van AI-libraries. Deze libraries zijn vaak alleen beschikbaar voor Python, waardoor het gebruik van Python een vereiste wordt, of zij zijn wel beschikbaar voor andere talen, maar worden niet officiëel ondersteund.

Naast de beschikbare libraries wordt Python gezien als een prettige taal voor AI-experimenten, vanwege de flexibiliteit en eenvoudige leercurve. Voor het ontwikkelen van een productiewaardig systeem zijn deze argumenten minder relevant, waardoor in volwassen projecten Python minder vaak gebruikt wordt. Bij een volwassen systeem worden andere argumenten, zoals stabiliteit en onderhoudbaarheid, juist belangrijker.

Met deze verschillen moet omgegaan kunnen worden in een project die zowel productiewaardig moet zijn als gebruik moet maken van AI. Hiervoor stellen wij het opsplitsen van de applicatie voor. Een deel van de applicatie, geschreven in Python, is verantwoordelijk voor AI en voert alleen die taak uit. Andere delen, geschreven in één of meerdere andere programmeertalen (bijvoorbeeld Kotlin), zijn verantwoordelijk voor alle andere logica. Hiervoor kan bijvoorbeeld een microservice architectuur gebruikt worden, waardoor je voor iedere service de programmeertaal die past bij de verantwoordelijkheden kunt kiezen. De delen van de applicatie communiceren met behulp van RabbitMQ.

Aan de hand van een casus en codevoorbeelden neem ik je mee door de implementatie van een applicatie die uit meerdere services bestaat, die onderling communiceren met behulp van RabbitMQ. De codevoorbeelden zijn terug te vinden in het project op GitHub, waarvoor de link en uitleg onderaan de blog te vinden zijn. Zo kun je zelf meedoen bij het implementeren van een applicatie, door onze casus of een andere casus naar keuze uit te werken!

Casus

De casus die ik je in deze blog laat zien gaat over een brouwerij die meerdere soorten bier kan brouwen. De gebruiker moet een foto van een biertje aan kunnen bieden, waarna de brouwerij het bier op de foto gaat brouwen.

Het brouwproces bestaat uit een aantal stappen:

• Bepalen om welk bier het gaat door middel van een image classification AI-model (Python, beer machine)
• Juiste ingrediënten regelen (Kotlin, inventory service)
• Bier brouwen (Kotlin, brewery service)
• Bier aan gebruiker leveren (Python, beer machine)

Tussen deze stappen moeten berichten gestuurd worden om de volgende stap te activeren. Berichten moeten dus van Python naar Kotlin en andersom. Hiervoor wordt RabbitMQ ingezet. In onderstaand diagram is te zien hoe de gebruiker alleen interactie heeft met de beer machine, waarna het proces langs de andere services loopt.

Het gebruik van RabbitMQ

RabbitMQ is een open source message broker die we als voorbeeld gebruiken in deze blog, je zou ook elke andere message broker kunnen gebruiken. Door de volwassenheid en vele functies gebruiken we RabbitMQ bij onze projecten en ook in deze blog. In deze blog gebruik ik vooral de basisfuncties die RabbitMQ biedt. De geavanceerdere functies worden in een toekomstige blog toegelicht.

Bij het gebruik van RabbitMQ zijn enkele begrippen van belang:

• Exchange: de plek waar berichten naartoe gestuurd worden en die de berichten plaatst in de juiste queue(s).
• Queue: een verzameling berichten in een bepaalde volgorde. Berichten die als eerste in de queue worden geplaatst zullen ook als eerste door de ontvanger uitgelezen worden. Dit wordt ook wel FIFO (first in, first out) genoemd.
• Routing key: een sleutel in de vorm van een string die als "adres" gebruikt wordt en door de ontvanger aan een queue gebonden wordt.

In onderstaande figuur is een voorbeeld te zien van hoe twee services communiceren via RabbitMQ. De Inventory service stuurt een bericht naar de routing key "brewing.requested". Die service hoeft niet te weten wat er daarna mee gebeurt, omdat RabbitMQ de aflevering van het bericht verzorgt. De brewery service registreert een queue waarop de berichten van die routing key geplaatst zullen worden. Hierdoor ontvangt de brewery die berichten van de inventory service.

Configuratie

Om deze berichten via RabbitMQ te kunnen versturen en ontvangen is één stuk configuratie nodig: een gebruiker die daarvoor de rechten heeft. Deze configuratie ziet er als volgt uit en geeft de gebruiker (beer) volledige rechten. Het is mogelijk en zelfs aan te raden om meerdere gebruikers aan te maken met precies de juiste rechten voor de service die het account gebruikt.

{
  "rabbit_version": "3.8.17",
  "rabbitmq_version": "3.8.17",
  "users": [
    {
      "name": "beer",
       "password_hash": "h5iCNjHhQLxHkGA/28MifZnDyO7aaJdR66jkXxem2BiNmoCG",
       "hashing_algorithm": "rabbit_password_hashing_sha256",
       "tags": "administrator"
    }
  ],
  "vhosts": [
    {
       "name": "/"
    }
  ],
  "permissions": [
    {
       "user": "beer",
       "vhost": "/",
       "configure": ".*",
       "write": ".*",
       "read": ".*"
    }
  ]
}

@Configuration
class MessagingConfiguration {

    companion object {
        const val EXCHANGE_NAME = "beer-exchange"
        const val QUEUE_NAME = "brewing-requests"
        const val BREWING_REQUEST_ROUTING_KEY = "brewing.requested"
        const val CORRELATION_HEADER_KEY = "correlation-id"
    }

    @Bean
    fun exchange() = DirectExchange(EXCHANGE_NAME)

    @Bean
    fun brewingRequestQueue(): Queue = QueueBuilder.durable(QUEUE_NAME).build()

    @Bean
    fun brewingRequestBinding(brewingRequestQueue: Queue, exchange: DirectExchange): Binding =
        BindingBuilder
            .bind(brewingRequestQueue)
            .to(exchange)
            .with(BREWING_REQUEST_ROUTING_KEY)

    @Bean
    fun messageSender(
        rabbitTemplate: RabbitTemplate,
        @Value("\${spring.application.name}") applicationName: String
    ) = MessageSender(rabbitTemplate, applicationName)

    @Bean
    fun rabbitTemplate(
        connectionFactory: ConnectionFactory,
        jackson2JsonMessageConverter: Jackson2JsonMessageConverter
    ): RabbitTemplate {
        val rabbitTemplate = RabbitTemplate(connectionFactory)
        rabbitTemplate.messageConverter = jackson2JsonMessageConverter
        rabbitTemplate.setMandatory(true)
        return rabbitTemplate
    }

    @Bean
    fun objectMapperBuilder(): Jackson2ObjectMapperBuilder =
        Jackson2ObjectMapperBuilder().modulesToInstall(KotlinModule())

    @Bean
    fun producerJackson2MessageConverter(objectMapperBuilder: Jackson2ObjectMapperBuilder): Jackson2JsonMessageConverter =
        Jackson2JsonMessageConverter(objectMapperBuilder.build())
}

spring:
  application:
    name: brewery
  rabbitmq:
    host: rabbitmq
    username: beer
    password: brewing

PERSISTENT_MODE: int = 2
CORRELATION_ID_HEADER_KEY = "BeerCorrelationId"


def get_connection() -> BlockingConnection:
    return BlockingConnection(
        ConnectionParameters(
            host=os.getenv("RABBITMQ_HOST"),
            credentials=PlainCredentials(
                username=os.getenv("RABBITMQ_USERNAME"),
                password=os.getenv("RABBITMQ_PASSWORD")
            )
        )
    )


def exchange_declare(channel: BlockingChannel, exchange_name: str, exchange_type: str = "direct") -> None:
    channel.exchange_declare(
        exchange=exchange_name,
        exchange_type=exchange_type,
        durable=True
    )


def queue_declare(channel: BlockingChannel, queue_name: str = "") -> Method:
    return channel.queue_declare(
        queue=queue_name,
        durable=True,
        arguments={'x-queue-type': 'quorum'}
    )


def queue_bind(channel: BlockingChannel, exchange_name: str, queue_name: str, routing_key: str) -> None:
    channel.queue_bind(
        exchange=exchange_name,
        queue=queue_name,
        routing_key=routing_key
    )


def basic_publish(channel: BlockingChannel, exchange_name: str, routing_key: str, message: BaseMessage) -> None:
    body: str = message.to_json()
    correlation_id: str = get_correlation_id()

    properties = BasicProperties(
        delivery_mode=PERSISTENT_MODE,
        headers={CORRELATION_ID_HEADER_KEY: correlation_id}
    )

    channel.basic_publish(
        exchange=exchange_name,
        routing_key=routing_key,
        body=bytes(body, "utf-8"),
        properties=properties,
        mandatory=True
    )


def basic_consume(channel: BlockingChannel, queue_name: str, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]) -> None:
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=callback
    )
    channel.start_consuming()

De juiste ingrediënten verzamelen

Het verstuurde bericht wordt in Kotlin ontvangen, in de inventory service. Dit komt doordat deze service de routing key gebonden heeft aan een queue en naar berichten op deze queue luistert. Deze service gaat op basis van het ontvangen soort bier de juiste ingrediënten bestellen.

Het ontvangen van het bericht gebeurt via een message receiver:

Message receiver in Kotlin 

@Component
class OrderRequestMessageReceiver(
    private val orderService: OrderService,
    private val messageSender: MessageSender
) {
    companion object : KLogging() {
        private const val BEER_BREWING_REQUEST_ROUTING_KEY = "brewing.requested"
    }

    @RabbitListener(queues = [MessagingConfiguration.QUEUE_NAME])
    fun receiveMessage(receivedMessage: OrderRequestedMessage, message: Message, channel: Channel) =
        processMessage(receivedMessage, message, channel) {
            val requestBrewingMessage = orderService.processOrderRequest(receivedMessage.beer)
            messageSender.sendMessage(requestBrewingMessage, BEER_BREWING_REQUEST_ROUTING_KEY)
        }

    protected fun processMessage(receivedMessage: OrderRequestedMessage, message: Message, channel: Channel, handleFunction: (OrderRequestedMessage) -> Unit) =
        withCorrelationId(message.messageProperties.headers[MessagingConfiguration.CORRELATION_HEADER_KEY]?.toString()) {
            logger.info("Received ${receivedMessage::class.simpleName} in ${this::class.simpleName}")
            handleFunction(receivedMessage)
            channel.basicAck(message.messageProperties.deliveryTag, false)
        }
}

class MessageSender(
    private val rabbitTemplate: RabbitTemplate,
    private val applicationName: String
) {
    companion object: KLogging()

    fun sendMessage(message: ServeBrewingMessage, routingKey: String) {
        logger.info("Sending message to [$routingKey] from $applicationName")
        rabbitTemplate.convertAndSend(EXCHANGE_NAME, routingKey, message, this::addCorrelationIdToMessage)
    }

    private fun addCorrelationIdToMessage(message: Message): Message =
        message.apply {
            messageProperties.setHeader(MessagingConfiguration.CORRELATION_HEADER_KEY, CorrelationHelper.getCorrelationId())
        }
}

def start_receiving(exchange_name: str, queue_name: str, routing_key: str, callback: Callable[[BlockingChannel, spec.Basic.Deliver, spec.BasicProperties, bytes], None]):
    connection: BlockingConnection = rabbit_connector.get_connection()
    channel: BlockingChannel = connection.channel()

    rabbit_connector.exchange_declare(
        channel=channel,
        exchange_name=exchange_name,
        exchange_type="direct"
    )

    rabbit_connector.queue_declare(
        channel=channel,
        queue_name=queue_name
    )

    rabbit_connector.queue_bind(
        channel=channel,
        exchange_name=exchange_name,
        queue_name=queue_name,
        routing_key=routing_key
    )

    rabbit_connector.basic_consume(
        channel=channel,
        queue_name=queue_name,
        callback=callback
    )

def __received_served_brewing_callback(channel: BlockingChannel, method: spec.Basic.Deliver, properties: spec.BasicProperties, body: bytes):
    message: ServeBrewingMessage = ServeBrewingMessage.from_json(body)

    correlation_id: str = properties.headers.get(CORRELATION_ID_HEADER_KEY)
    set_or_generate_correlation_id(correlation_id)

    print("Served a beer: ", message.beer)
    print("The brewing process was completed in " + str(message.brewing_time) + " seconds.")

    channel.basic_ack(delivery_tag=method.delivery_tag)