[rabbitmq] 安装与配置

RabbitMQ
-------------------------
1
下载

erlang
http://www.erlang.org/download/otp_win32_17.3.exe
http://www.erlang.org/download/otp_src_17.3.tar.gz
http://www.erlang.org/download/otp_doc_html_17.3.tar.gz
http://www.erlang.org/download/otp_doc_man_17.3.tar.gz

rabbitmq server
http://www.rabbitmq.com/releases/rabbitmq-server/v3.4.2/rabbitmq-server-3.4.2.exe

rabbitmq java client
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-bin-3.4.2.zip
http://www.rabbitmq.com/releases/rabbitmq-java-client/v3.4.2/rabbitmq-java-client-3.4.2.zip

2
配置

3
基本概念
http://www.rabbitmq.com/tutorials/tutorial-one-java.html

RabbitMQ 是一个消息代理,它接收生产者的消息,然后转发给消费者.同时,它路由,缓存,持久这个消息根据设定的规则.
producer 生产者,只负责发送消息.消息发送到指定队列中.
queue 队列或邮箱,它存储消息,它的大小没有限制,只要磁盘空间足够.消息只能缓存在队列中,队列属于消息代理的一部分.消息代理还包括 exchange 等.
consumer 消费者,它从队列接收消息.

AMQP:高级消息队列协议.

P(producer)   -- msg --> |Queue|Queue|Queue|  -- msg --> C(consumer)

PS:
生产者,消息代理,消费者通常不在同一机器上.

4 Hello World
----------------
我们这里使用 Java 客户端.

STEP 1:

    P(producer)   -- msg --> |HelloQueue|

public class P {

  private final static String QUEUE_NAME = "HelloQueue";

  public static void main(String[] args)
      throws Exception {
      // 建立连接
      ConnectionFactory factory = new ConnectionFactory();
      factory.setHost("127.0.0.1"); 
      Connection connection = factory.newConnection();
      Channel channel = connection.createChannel();
      
      // 声明队列
      channel.queueDeclare(QUEUE_NAME, false, false, false, null);
      
      // 创建消息
      String message = "Hello World!";
      
      // 发布消息到 HelloQueue
      channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
      
      // 关闭连接
      channel.close();
      connection.close();
  }

}

STEP 2

    |HelloQueue| -- msg --> C(consumer)

public class C {

  private final static String QUEUE_NAME = "HelloQueue";

  public static void main(String[] args)
      throws Exception {
    
    // 建立连接
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();
    
    // 声明队列
    channel.queueDeclare(QUEUE_NAME, false, false, false, null);
    
    // 创建消费者,处理消息
    QueueingConsumer consumer = new QueueingConsumer(channel);
    channel.basicConsume(QUEUE_NAME, true, consumer);

    while (true) {
        // block method ? yes.
        // 阻塞直到接收到下一条消息
        QueueingConsumer.Delivery delivery = consumer.nextDelivery(); 
        String message = new String(delivery.getBody());
        System.out.println("Received : " + message);
    }
}
  

5
Work Queue

    P -- MSG -- |Q| -- MSG -- Worker1|Worker2

Work1 , Work2 轮流获取消息处理,
为避免Worker1消息单节点处理失败, 修改ACK返回时机为消息处理成功后.ACK返回,消息代理才会删除已处理的消息,否则是Worker1一接收到就删除.

    boolean autoAck = false;
    channel.basicConsume("hello", autoAck, consumer);

while (true) {
    QueueingConsumer.Delivery delivery = consumer.nextDelivery();
    //...      
    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}

为保证消息代理节点失败,导致消息丢失, 设置消息持久化参数:

boolean durable = true;
channel.queueDeclare("hello", durable, false, false, null);

...

 channel.basicPublish( "", TASK_QUEUE_NAME, 
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());

6
发布/订阅模式
一个消息多个接收者接收.
EXCHANGE: 消息交换,用于路由/过滤消息.

P -- MSG -- |X| -- |Q1| -- MSG -- C1
                -- |Q2| -- MSG -- C2

     // 交换策略扇出(fanout):消息交换到所有队列 
     channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

     String message = getMessage(argv);

     channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
     ...

     channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
     // 临时队列.随机命名的队列
     String queueName = channel.queueDeclare().getQueue();
     // 绑定交换与队列
     channel.queueBind(queueName, EXCHANGE_NAME, "");

7
路由
绑定: 一个队列对交换的消息感兴趣.
绑定KEY: 绑定时设置
路由KEY: 发布时设置 routingKey, fanout 会简单忽略这个值.
交换策略:
direct exchange: 直接交换,根据 routingKey 来选择发布到哪些队列.比fanout更加灵活.
fanout exchange: 扇出交换(全部队列都会接收这个消息)
topic exchange: routingKey用 "." 号分开.根据 routingKey 来选择发布到哪些匹配的队列, 支持模糊匹配.

     channel.exchangeDeclare(EXCHANGE_NAME, "topic");
     ...
     channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());

      channel.exchangeDeclare(EXCHANGE_NAME, "topic");
      String queueName = channel.queueDeclare().getQueue();



      for(String bindingKey : bindingKeys){
          channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
      }

      QueueingConsumer consumer = new QueueingConsumer(channel);
      channel.basicConsume(queueName, true, consumer);

      while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
          String routingKey = delivery.getEnvelope().getRoutingKey();
          ...
        }

8
RPC 远程过程调用
通常发送方要等待接收方的处理结果.

发送方阻塞直到方法返回结, 每个客户端一个响应队列.
为了识别哪个响应属于哪个请求,要加个请求ID.

result = Client.call(request); // block until result has returned

    Client -- id:RequestId , replyTo:ReplyQueue -- HelloQueue -- Server
           |                                                       |
           -- id:RequestId , replyTo:ReplyQueue -- ReplyQueue ------

RPCServer

private static final String RPC_QUEUE_NAME = "rpc_queue";

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

channel.basicQos(1);

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

System.out.println(" [x] Awaiting RPC requests");

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

    BasicProperties props = delivery.getProperties();
    BasicProperties replyProps = new BasicProperties
                                     .Builder()
                                     .correlationId(props.getCorrelationId())
                                     .build();

    String message = new String(delivery.getBody());
    int n = Integer.parseInt(message);

    System.out.println(" [.] fib(" + message + ")");
    String response = "" + fib(n);

    channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes());

    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}

RPCClient

private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
private String replyQueueName;
private QueueingConsumer consumer;

public RPCClient() throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    connection = factory.newConnection();
    channel = connection.createChannel();

    replyQueueName = channel.queueDeclare().getQueue(); 
    consumer = new QueueingConsumer(channel);
    channel.basicConsume(replyQueueName, true, consumer);
}

public String call(String message) throws Exception {     
    String response = null;
    String corrId = java.util.UUID.randomUUID().toString();

    BasicProperties props = new BasicProperties
                                .Builder()
                                .correlationId(corrId)
                                .replyTo(replyQueueName)
                                .build();

    channel.basicPublish("", requestQueueName, props, message.getBytes());

    while (true) {
        QueueingConsumer.Delivery delivery = consumer.nextDelivery();
        if (delivery.getProperties().getCorrelationId().equals(corrId)) {
            response = new String(delivery.getBody());
            break;
        }
    }

    return response; 
}

public void close() throws Exception {
    connection.close();
}