gistfile1.scala

import akka.amqp.AMQP._
import akka.amqp._
import akka.actor._
import akka.util.Logging

object LoadBalancingTest extends Logging {

    val connection = AMQP.newConnection()
    val exchangeParameters = ExchangeParameters("my_direct_exchange", Direct)

    class JobConsumer(id: Int) extends Actor {

        def receive = {
            case Delivery(payload, _, _, _, _, _) =>
                println("(" + id + ") received message: " + new String(payload))
        }
    }

    def main(args: Array[String]) {
        // == Consumer
        val channelParameters = Some(ChannelParameters(prefetchSize = 1))
        for (i <- 1 to 3) {
            val actor = Actor.actorOf(new JobConsumer(i))
            AMQP.newConsumer(connection, ConsumerParameters("some.routing.key",
                actor, Some("my-job-queue"), Some(exchangeParameters), channelParameters = channelParameters))
        }

        // == Producer
        val producer = AMQP.newProducer(connection,
            ProducerParameters(Some(exchangeParameters), Some("my-producer")))

      for (i <- 1 to 30) {
        producer ! Message(("data (" + i + ")").getBytes, "some.routing.key")
        Thread.sleep(200)
      }
    }
}