RabbitMQ

Overview

RabbitMQ digunakan sebagai message bus untuk komunikasi antar services.

Configuration

Docker Compose

rabbitmq:
  image: rabbitmq:3-management
  container_name: rabbitmq
  environment:
    RABBITMQ_DEFAULT_USER: admin
    RABBITMQ_DEFAULT_PASS: rabbitmq123
  ports:
    - "5672:5672"
    - "15672:15672"
  volumes:
    - rabbitmq-data:/var/lib/rabbitmq
    - ./configs/rabbitmq/definitions.json:/etc/rabbitmq/definitions.json:ro
    - ./configs/rabbitmq/init.sh:/docker-entrypoint-initdb.d/init.sh:ro
  mem_limit: 256m
  cpus: "0.25"
  healthcheck:
    test: ["CMD", "rabbitmq-diagnostics", "check_port_connectivity"]
    interval: 30s
    timeout: 10s
    retries: 3
    start_period: 30s

definitions.json

{
  "exchanges": [
    {
      "name": "clm.events",
      "vhost": "/",
      "type": "topic",
      "durable": true,
      "auto_delete": false
    }
  ],
  "queues": [
    {
      "name": "clm.alerts",
      "vhost": "/",
      "durable": true,
      "auto_delete": false
    },
    {
      "name": "clm.audit",
      "vhost": "/",
      "durable": true,
      "auto_delete": false
    },
    {
      "name": "clm.notifications",
      "vhost": "/",
      "durable": true,
      "auto_delete": false
    }
  ],
  "bindings": [
    {
      "source": "clm.events",
      "vhost": "/",
      "destination": "clm.alerts",
      "destination_type": "queue",
      "routing_key": "alert.*"
    },
    {
      "source": "clm.events",
      "vhost": "/",
      "destination": "clm.audit",
      "destination_type": "queue",
      "routing_key": "audit.*"
    },
    {
      "source": "clm.events",
      "vhost": "/",
      "destination": "clm.notifications",
      "destination_type": "queue",
      "routing_key": "notification.*"
    }
  ]
}

Exchanges & Queues

Exchanges

Exchange	Type	Description
clm.events	topic	Main event exchange

Queues

Queue	Routing Key	Description
clm.alerts	alert.*	Alert events
clm.audit	audit.*	Audit events
clm.notifications	notification.*	Notification events

Publishing Messages

Go Backend

import "github.com/streadway/amqp"

func PublishAlert(conn *amqp.Connection, alert Alert) error {
    ch, err := conn.Channel()
    if err != nil {
        return err
    }
    defer ch.Close()
    
    body, _ := json.Marshal(alert)
    
    return ch.Publish(
        "clm.events",
        "alert.created",
        false,
        false,
        amqp.Publishing{
            ContentType: "application/json",
            Body:        body,
        },
    )
}

Consumer Service

func ConsumeAlerts(conn *amqp.Connection) error {
    ch, err := conn.Channel()
    if err != nil {
        return err
    }
    defer ch.Close()
    
    msgs, err := ch.Consume(
        "clm.alerts",
        "",
        true,
        false,
        false,
        false,
        nil,
    )
    
    for msg := range msgs {
        var alert Alert
        json.Unmarshal(msg.Body, &alert)
        processAlert(alert)
    }
    
    return nil
}

Management UI

Access

http://localhost:15672

Credentials

Username	Password
admin	rabbitmq123

CLI Commands

# List queues
docker exec rabbitmq rabbitmqctl list_queues

# List connections
docker exec rabbitmq rabbitmqctl list_connections

# List channels
docker exec rabbitmq rabbitmqctl list_channels

Monitoring

Health Check

docker exec rabbitmq rabbitmq-diagnostics check_port_connectivity

Metrics

curl -u admin:rabbitmq123 http://localhost:15672/api/overview

Troubleshooting

Check Logs

docker logs rabbitmq --tail 100

Restart Service

docker compose restart rabbitmq

Next Steps

• Consumer Service - Message consumer
• Data Pipeline - Pipeline overview