RabbitMQ指南(上)

RabbitMQ是一个消息中间件,在一些需要异步处理、发布/订阅等场景的时候,使用RabbitMQ可以完成我们的需求。 下面是我在学习RabbitMQ的过程中的一些记录,内容主要翻译自RabbitMQ官网的Tutorials, 再加上我的一些个人理解。我将会用三篇文章来从RabbitMQ的Hello World介绍起,到最后的通过RabbitMQ实现RPC调用, 相信看完这三篇文章大家应该会对RabbitMQ的基本概念和使用有一定的了解。

说明:

  1. 由于RabbitMQ支持许多种语言的client,在这里我使用的是Java语言的client。
  2. 所有的图片均来自RabbitMQ官网。

Hello World

首先需要安装RabbitMQ,关于RabbitMQ的安装这里就不赘述了,可以到RabbitMQ的官网去看相应的OS的安装方法。 安装完成后使用rabbitmq-server即可启动RabbitMQ,RabbitMQ还提供了一个UI管理界面,本地默认的地址为localhost:15672, 用户名和密码均为guest。

安装完成之后,按照惯例,先来完成一个简单的Hello World的例子。 最简单的一种消息发送的模型为一个消息发送者(Producer)将消息发送到Queue中,另一端的消息接受者(Consumer)从Queue中接受消息, 大致模型如下图所示:

RabbitMQ

先来看发送的代码,新建一个类命名为Send.java,代码的第一步为连接server

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

connection抽象了socket的连接,并且为我们处理了协议版本的协商、权限认证等等。这里我们连接的是本地的中间件, 也就是localhost,接下来我们创建一个channel,这是大多数API完成任务的所在,也就是说我们的API操作基本都是通过channel来完成的。

channel.queueDeclare(QUEUE_NAME, false, false, false, null);
String message = "Hello World!";
channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");

首先是通过channel来声明一个queue,并且声明queue的操作是幂等的,也即是说只有在这个queue不存在的情况下才会新创建一个queue。 这里发送一个Hello World!的消息,实际传递的消息内容为字节数组。

channel.close();
connection.close();

最后关闭channel和connection的连接,注意关闭的顺序,是先关闭channel的连接,再关闭connection的连接。

完整的Send.java代码

public class Send {

    private static final String QUEUE_NAME = "hello";

    public static void main(String[] args) {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.queueDeclare(QUEUE_NAME, false, false, false, null);
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");

        channel.close();
        connection.close();
    }

}

完成发送的代码之后是接受消息的代码,新建一个类为Recv.java

public class Recv {

    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) 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);
      System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

      Consumer consumer = new DefaultConsumer(channel) {
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body)
            throws IOException {
          String message = new String(body, "UTF-8");
          System.out.println(" [x] Received '" + message + "'");
        }
      };
      channel.basicConsume(QUEUE_NAME, true, consumer);
    }
}

可以发现一开始的连接部分的代码是相同的,在接收的时候我们也要声明一个queue,注意这里queue的名称和之前发送消息声明的queue的名称必须是相同的, 否则就收不到消息了。

DefaultConsumer类实现了Consumer接口,由于发送消息是异步的,因此在这里提供了一个callback来缓冲消息, 直到我们准备使用这些消息,最后分别运行Send.javaRecv.java,就能看到Hello World!消息了。

Work Queues

在第一部分的Hello World中通过一个命名的queue来传递消息,在这一部分,我们会创建Work Queue来将耗时的任务分发至多个worker。 假设一个消息就是一个耗时的任务,比如文件I/O等等,那么可以通过几个worker来共同完成这些工作。

RabbitMQ

在Web应用中这是非常有用的,因为在一次非常短的HTTP请求窗口中完成一个非常复杂的任务是很困难的。

准备

这一部分是建立在上一部分Hello World的基础之上的,我们将发送字符串来表示一些复杂的任务, 由于并没有一些真实的复杂的工作,因此使用Thread.sleep()来模拟这是一个很耗时的任务, 并且在发送的字符串当中含有一个点号就表示这个任务需要耗时1秒,比如发送Hello...表示将要耗时3秒。

在前一部分的Send.java的基础上做一些修改,得到一个新的类称为NewTask.java

String message = getMessage(argv);

channel.basicPublish("", "hello", null, message.getBytes());
System.out.println(" [x] Sent '" + message + "'");

getMessage方法为从命令行中获取参数

private static String getMessage(String[] strings){
    if (strings.length < 1)
        return "Hello World!";
    return joinStrings(strings, " ");
}

private static String joinStrings(String[] strings, String delimiter) {
    int length = strings.length;
    if (length == 0) return "";
    StringBuilder words = new StringBuilder(strings[0]);
    for (int i = 1; i < length; i++) {
        words.append(delimiter).append(strings[i]);
    }
    return words.toString();
}

我们之前的Recv.java也需要做一些变化,它需要模拟一些耗时的任务,消息内容中一个.表示1秒,并且它会处理消息, 我们称它为Worker.java

final Consumer consumer = new DefaultConsumer(channel) {
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    String message = new String(body, "UTF-8");

    System.out.println(" [x] Received '" + message + "'");
    try {
      doWork(message);
    } finally {
      System.out.println(" [x] Done");
    }
  }
};
boolean autoAck = true; // acknowledgment is covered below
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

这里有一个autoAck变量的作用在后面会提到。doWork方法就是模拟的耗时任务

private static void doWork(String task) throws InterruptedException {
    for (char ch: task.toCharArray()) {
        if (ch == '.') Thread.sleep(1000);
    }
}

循环发送

使用任务队列其中的一个好处是可以非常方便的并行处理这些任务。如果我们在处理一些积压的工作, 只需要增加更多的worker即可,非常容易扩展。

首先,来试试两个worker实例的情况。很显然,两个worker都会接受到消息,但是具体的情况是怎么样的呢? 我们在控制台启动两个实例,C1和C2表示两个consumer,然后使用Producer来发送消息,一共发送五条消息,来看看具体的情况。

首先是第一个worker打印出的消息

[*] Waiting for messages. To exit press CTRL+C
[x] Received 'First message.'
[x] Received 'Third message...'
[x] Received 'Fifth message.....'

第二个worker打印出的消息

[*] Waiting for messages. To exit press CTRL+C
[x] Received 'Second message..'
[x] Received 'Fourth message....'

默认的,RabbitMQ会顺序的把消息发送到下一个Consumer,上面打印出的消息也印证了这一点。 平均来说每个Consumer接收到的消息数量是相同的,这种发送消息的方式称为循环发送(round-robin), 思考下有三个或者更多Worker的情况。

消息接收(Message acknowledgment)

处理一个任务需要耗费几秒钟的时间。你也许想知道如果一个consumer在处理一个任务的时候只处理了一部分就挂了会出现什么情况。 在我们现在的代码下,一旦RabbitMQ将一个消息传递到consumer,它马上会从内存中删除这条消息, 也就是说如果杀掉了一个正在处理任务的worker,那么将会失去所有的这个worker正在处理的所有消息, 同样也会失去发送给这个worker但是还未处理的消息。

一般情况下,我们不希望丢失消息,如果某个worker挂了,能将消息发送给另一个worker来处理。 为了确保消息不会丢失,RabbitMQ支持消息接收(message acknowledgments)。 当consumer确认收到某个消息,并且已经处理完成,RabbitMQ可以删除它时,consumer会向RabbitMQ发送一个ack(nowledgement)

如果一个consumer挂了(channel关闭了、connection关闭了或者TCP连接断了)而没有发送ack,RabbitMQ就会知道这个消息没有被完全处理, 将会对这条消息做re-queue处理。如果此时有另一个consumer连接,消息会被重新发送至另一个consumer。 使用这种方式可以保证消息不会丢失。

消息不会超时;RabbitMQ会在consumer挂了之后重新发送消息。即使处理消息耗时非常长也是没有问题的。

消息接收是默认开启的,在之前的例子中我们通过autoAck=true标志显式的关闭了它,true则表示自动接收,不需要发送ack。 现在是时候来开启ack。当consumer处理完成之后,向rabbitMQ发送ack。

channel.basicQos(1); // accept only one unack-ed message at a time (see below)

final Consumer consumer = new DefaultConsumer(channel) {
  @Override
  public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    String message = new String(body, "UTF-8");

    System.out.println(" [x] Received '" + message + "'");
    try {
      doWork(message);
    } finally {
      System.out.println(" [x] Done");
      channel.basicAck(envelope.getDeliveryTag(), false);
    }
  }
};
boolean autoAck = false;
channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);

使用以上代码能够保证即使在一个worker处使消息的时候用CTRL+C来杀掉这个worker,也不会丢失消息。 在这个worker挂掉之后所有未接收(ack)的消息将被重新发送。

消息持久化

我们学习了如何在worker挂掉的情况下不丢消息,但是在RabbitMQ server停止之后消息还是会丢失。 如果不进行任何配置,在RabbitMQ退出或崩溃的时候,将会失去所有的queue和消息。 要保证在这种情况下消息不丢失需要做两件事情:需要同时标志queue和message是持久化的。

首先,需要确保RabbitMQ不会丢失queue

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

我们重新声明一个queue(不能修改已经声明为不持久化的queue为持久化),名字为task_queue, 第二个布尔参数表示是否持久化的意思,这里设置为true, 包括consumer和producer声明queue的时候都需要声明durable为true。现在,即使重启RabbitMQ,task_queue这个queue也不会丢失了。

接下来我们将消息做持久化配置处理,通过设置MessageProperties(实现了BasicProperties)中的PERSISTENT_TEXT_PLAIN属性。

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

公平分发(Fair dispatch)

在某种场景下有两个worker,当所有奇数的消息处理起来都比较耗时,而偶数的消息处理起来都比较快, 这就会发生一个worker总是处于busy状态,而另一个worker则总是处于空闲状态,RabbitMQ并不知道这个情况, 仍然只是正常的发送消息。

出现这种情况的原因在于当消息在queue中的时候RabbitMQ只是发送这些消息而已,它不会去关注某个consumer未ack的消息的数量, 它只是盲目的将某个消息发送到某个consumer。

RabbitMQ

为了处理这种情况我们可以使用basicQos方法来设置prefetchCount = 1。 这告诉RabbitMQy一次只给worker一条消息,换句话来说,就是直到worker发回ack,然后再向这个worker发送下一条消息。

int prefetchCount = 1;
channel.basicQos(prefetchCount);

完整的NewTask.java代码

public class NewTask {

  private static final String TASK_QUEUE_NAME = "task_queue";

  public static void main(String[] argv)
                      throws java.io.IOException {

    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    Connection connection = factory.newConnection();
    Channel channel = connection.createChannel();

    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

    String message = getMessage(argv);

    channel.basicPublish( "", TASK_QUEUE_NAME,
            MessageProperties.PERSISTENT_TEXT_PLAIN,
            message.getBytes());
    System.out.println(" [x] Sent '" + message + "'");

    channel.close();
    connection.close();
  }      
  //...
}

Worker.java

public class Worker {
  private static final String TASK_QUEUE_NAME = "task_queue";

  public static void main(String[] argv) throws Exception {
    ConnectionFactory factory = new ConnectionFactory();
    factory.setHost("localhost");
    final Connection connection = factory.newConnection();
    final Channel channel = connection.createChannel();

    channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
    System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

    channel.basicQos(1);

    final Consumer consumer = new DefaultConsumer(channel) {
      @Override
      public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");

        System.out.println(" [x] Received '" + message + "'");
        try {
          doWork(message);
        } finally {
          System.out.println(" [x] Done");
          channel.basicAck(envelope.getDeliveryTag(), false);
        }
      }
    };
    boolean autoAck = false;
    channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
  }

  private static void doWork(String task) {
    for (char ch : task.toCharArray()) {
      if (ch == '.') {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException _ignored) {
          Thread.currentThread().interrupt();
        }
      }
    }
  }
}

可以运行上面两个类来验证这一小节的所有内容。



相关文章

发表评论

Comment form

(*) 表示必填项

还没有评论。

跳到底部
返回顶部