RabbitMQ(9)Have Working Queue on MAC with Scala in Cluster Mode

RabbitMQ(9)Have Working Queue on MAC with Scala in Cluster Mode

1. Install on my MAC laptop.
>su root
>port selfupdate
>sudo port install erlang +ssl

Download the source from URL
http://www.rabbitmq.com/releases/rabbitmq-server/v3.1.5/rabbitmq-server-generic-unix-3.1.5.tar.gz

>tar zxvf rabbitmq-server-generic-unix-3.1.5.tar.gz
>mv rabbitmq_server-3.1.5 /Users/carl/tool/
>sudo ln -s /Users/carl/tool/rabbitmq_server-3.1.5 /opt/rabbitmq-3.1.5
>sudo ln -s /opt/rabbitmq-3.1.5 /opt/rabbitmq

Start the Server
>sudo sbin/rabbitmq-server

Check and Stop the Server
>sudo sbin/rabbitmqctl status
>sudo sbin/rabbitmqctl stop

>sudo vi ~/.profile
export PATH=/opt/rabbitmq/sbin:$PATH
>. ~/.profile 

2. Example of Producer and Consumer
The sample project is in super duty .

Here is how we build the connection
package com.sillycat.superduty.service

import com.rabbitmq.client.ConnectionFactory
import com.rabbitmq.client.QueueingConsumer

object RabbitMQConnectionService {

  val TASK_QUEUE_NAME = "task_queue"

  val SERVER_HOST = "localhost"

  def connect = {
    val factory = new ConnectionFactory
    factory.setHost(SERVER_HOST)

    val conn = factory.newConnection()
    val channel = conn.createChannel()

    val durable = true
    channel.queueDeclare(TASK_QUEUE_NAME, durable, false, false, null)

    val prefetchCount = 1
    channel.basicQos(prefetchCount)

    val consumer = new QueueingConsumer(channel)
    val autoAck = false

    channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer)

    (conn, channel, consumer)
  }
}

Here is the Producer
package com.sillycat.superduty.jobs.producer

import com.sillycat.superduty.service.RabbitMQConnectionService
import com.rabbitmq.client.MessageProperties

object NewTaskRabbitMQ extends App {

  val (conn, channel, consumer) = RabbitMQConnectionService.connect

  val message = "campaign"

  Range(1, 10, 1).foreach { num =>
    channel.basicPublish("", RabbitMQConnectionService.TASK_QUEUE_NAME,
      MessageProperties.PERSISTENT_TEXT_PLAIN, (message + num).getBytes());
  }
  channel.close
  conn.close
}

Here is the Consumer

package com.sillycat.superduty.jobs.consumer

import com.sillycat.superduty.service.RabbitMQConnectionService
import com.rabbitmq.client.QueueingConsumer
import com.typesafe.scalalogging.slf4j.Logging

object WorkerRabbitMQ extends App with Logging {
  val (conn, channel, consumer) = RabbitMQConnectionService.connect

  while (true) {
    val delivery: QueueingConsumer.Delivery = consumer.nextDelivery()

    val message = new String(delivery.getBody())
    val deliverTag = delivery.getEnvelope.getDeliveryTag

    logger.debug("Worker get task=" + message + " delivery tag=" + deliverTag)
    channel.basicAck(deliverTag, false)
  }
}

Here is how we execute the command
>sudo sbin/rabbitmq-server


>sbt 'run-main com.sillycat.superduty.jobs.consumer.WorkerRabbitMQ'
>sbt 'run-main com.sillycat.superduty.jobs.consumer.WorkerRabbitMQ'


>sbt 'run-main com.sillycat.superduty.jobs.producer.NewTaskRabbitMQ'

And the build.sbt should add these dependencies.
        "com.rabbitmq"        %   "amqp-client"               % "3.1.4" 

3. Cluster Configuration
Should always follow this document
http://www.rabbitmq.com/clustering.html

>chmod 777 ~/.erlang.cookie
change the content of the file if you need
>chmod 400 ~/.erlang.cookie
>sudo chown -R carl ~/.erlang.cookie 
>sudo chgrp -R staff ~/.erlang.cookie

Enable the web UI management
>rabbitmq-plugins enable rabbitmq_management 

Start cluster on single machine
>RABBITMQ_NODE_PORT=5672 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15672}]" RABBITMQ_NODENAME=rabbit1 sbin/rabbitmq-server -detached
>RABBITMQ_NODE_PORT=5673 RABBITMQ_SERVER_START_ARGS="-rabbitmq_management listener [{port,15673}]" RABBITMQ_NODENAME=rabbit2 sbin/rabbitmq-server -detached

Check status of node1
>rabbitmqctl -n rabbit1 cluster_status
Cluster status of node rabbit1@sparkworker1 ...
[{nodes,[{disc,[rabbit1@sparkworker1]}]},
 {running_nodes,[rabbit1@sparkworker1]},
 {partitions,[]}]
…done.
 
>rabbitmqctl -n rabbit2 cluster_status 
Cluster status of node rabbit2@sparkworker1 ...
[{nodes,[{disc,[rabbit2@sparkworker1]}]},
 {running_nodes,[rabbit2@sparkworker1]},
 {partitions,[]}]
...done.

Stop the second node and make it join the first cluster
>rabbitmqctl -n rabbit2 stop_app
>rabbitmqctl -n rabbit2 join_cluster rabbit1@sparkworker1
>rabbitmqctl -n rabbit2 start_app

Check the status again
>rabbitmqctl -n rabbit1 cluster_status
Cluster status of node rabbit1@sparkworker1 ...
[{nodes,[{disc,[rabbit1@sparkworker1,rabbit2@sparkworker1]}]},
 {running_nodes,[rabbit2@sparkworker1,rabbit1@sparkworker1]},
 {partitions,[]}]
...done.

Visit the page
http://localhost:15672/

The default username and password should be guest/guest.

Changing the node Type
>rabbitmqctl -n rabbit2 stop_app
>rabbitmqctl -n rabbit2 change_cluster_node_type ram
>rabbitmqctl -n rabbit2 start_app

or 
>rabbitmqctl -n rabbit1 stop_app
>rabbitmqctl -n rabbit1 change_cluster_node_type disc
>rabbitmqctl -n rabbit1 start_app

The client will be something like this if connect to cluster
    val factory = new ConnectionFactory
    //factory.setHost(SERVER_HOST)
    //val conn = factory.newConnection()

    val addrArr = Array(new Address("localhost", 5672), new Address("", 5673))
    val conn = factory.newConnection(addrArr)

The source codes here is like this:
for (Address addr : addrs) {
            try {
                FrameHandler frameHandler = createFrameHandler(addr);
                AMQConnection conn =
                    new AMQConnection(username,
                                      password,
                                      frameHandler,
                                      executor,
                                      virtualHost,
                                      getClientProperties(),
                                      requestedFrameMax,
                                      requestedChannelMax,
                                      requestedHeartbeat,
                                      saslConfig);
                conn.start();
                return conn;
            } catch (IOException e) {
                lastException = e;
            }
        }

So, there is no load balance and we will try to get connection one by one till we get one, it is not good. We can consider put ha-proxy or nginx before that.

Break a Cluster
>rabbitmqctl -n rabbit2 stop_app
>rabbitmqctl -n rabbit2 reset 


References:
RabbitMQ 1 ~ 8
http://sillycat.iteye.com/blog/1565771
http://sillycat.iteye.com/blog/1567052
http://sillycat.iteye.com/blog/1575002
http://sillycat.iteye.com/blog/1575314
http://sillycat.iteye.com/blog/1575816
http://sillycat.iteye.com/blog/1578635
http://sillycat.iteye.com/blog/1579464
http://sillycat.iteye.com/blog/1582971
https://github.com/koeninger/rabbit-example

http://www.rabbitmq.com/install-macports.html
http://www.rabbitmq.com/getstarted.html

Message Working Queue
http://sillycat.iteye.com/blog/1575314
https://github.com/koeninger/rabbit-example

Cluster
http://www.rabbitmq.com/clustering.html
http://www.rabbitmq.com/partitions.html
http://www.rabbitmq.com/nettick.html

management page
http://www.rabbitmq.com/management.html

All the Client
http://www.rabbitmq.com/devtools.html

Tips 
http://fengchangjian.com/?p=1896