RabbitMQ指南（上）

栏目: Redis · 发布时间: 6年前

内容简介：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.java 和 Recv.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();
        }
      }
    }
  }
}

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


以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。

码农书籍
PHP Web 2.0开发实战

PHP Web 2.0开发实战

泽瓦斯 / 苏金国 / 人民邮电出版社 / 2008-10-1 / 59.00元

本书通过一个完整的Web 2.0应用——带有动态图库、搜索和地图功能的博客系统详细介绍了Web开发的全过程。首先讨论了Web应用的规划与设计，然后逐章实现各个具体特性，包括网站主页、用户主页、用户注册页面、账户登录和管理页面、用户博客系统、网站搜索以及应用管理等，最后介绍部署和维护。 本书适合中、高级的PHP程序员阅读。一起来看看 《PHP Web 2.0开发实战》 这本书的介绍吧!

码农工具
HTML 压缩/解压工具
HTML 压缩/解压工具

在线压缩/解压 HTML 代码

XML 在线格式化
XML 在线格式化

在线 XML 格式化压缩工具

HEX HSV 转换工具
HEX HSV 转换工具

HEX HSV 互换工具

  • New
  • 文章
  • 话题
  • 教程
    • 关注 码农网 公众号
    版权所有，保留一切权利！© 2018-2024 码农网 粤ICP备17054400号-3