RabbitMQ原理及实现

骑猪看日落 2022-06-08 09:38 241阅读 0赞

RabbitMQ的官方网站https://www.rabbitmq.com/

AMQP(Advanced Message Queuing Protocol),是一个提供统一消息服务的应用层标准高级消息队列协议,是应用层协议的一个开放标准,为面向消息的中间件设计。RabbitMQ就是这个协议的实现。

RabbitMQ在实现中包含下面几个组件：

1.Server(broker): 接受客户端连接，实现AMQP消息队列和路由功能的进程。

2.Virtual Host:其实是一个虚拟概念，类似于权限控制组，可以通过命令分配给用户Virtual Host的权限，默认的guest用户是管理员权限，初始空间有/，一个Virtual Host里面可以有若干个Exchange和Queue，但是权限控制的最小粒度是Virtual Host

3.Exchange:接受生产者发送的消息，并根据Binding规则将消息路由给服务器中的队列。ExchangeType决定了Exchange路由消息的行为，例如，在RabbitMQ中，ExchangeType有direct、Fanout和Topic三种，不同类型的Exchange路由的行为是不一样的。

4.Message Queue：消息队列，用于存储还未被消费者消费的消息。

5.Message: 由Header和Body组成，Header是由生产者添加的各种属性的集合，包括Message是否被持久化、由哪个Message Queue接受、优先级是多少等。而Body是真正需要传输的APP数据。

6.Binding:Binding联系了Exchange与Message Queue。Exchange在与多个Message Queue发生Binding后会生成一张路由表，路由表中存储着Message Queue所需消息的限制条件即Binding Key。当Exchange收到Message时会解析其Header得到Routing Key，Exchange根据Routing Key与Exchange Type将Message路由到Message Queue。Binding Key由Consumer在Binding Exchange与Message Queue时指定，而Routing Key由Producer发送Message时指定，两者的匹配方式由Exchange Type决定。

7.Connection:连接，对于RabbitMQ而言，其实就是一个位于客户端和Broker之间的TCP连接。

8.Channel:信道，仅仅创建了客户端到Broker之间的连接后，客户端还是不能发送消息的。需要为每一个Connection创建Channel，AMQP协议规定只有通过Channel才能执行AMQP的命令。一个Connection可以包含多个Channel。之所以需要Channel，是因为TCP连接的建立和释放都是十分昂贵的，如果一个客户端每一个线程都需要与Broker交互，如果每一个线程都建立一个TCP连接，暂且不考虑TCP连接是否浪费，就算操作系统也无法承受每秒建立如此多的TCP连接。RabbitMQ建议客户端线程之间不要共用Channel，至少要保证共用Channel的线程发送消息必须是串行的，但是建议尽量共用Connection。

9.Command:AMQP的命令，客户端通过Command完成与AMQP服务器的交互来实现自身的逻辑。例如在RabbitMQ中，客户端可以通过publish命令发送消息，txSelect开启一个事务，txCommit提交一个事务。

AMQP的三层协议

1.Module Layer，位于协议最高层，主要定义了一些供客户端调用的命令，客户端可以利用这些命令实现自己的业务逻辑，例如，客户端可以通过queue.declare声明一个队列，利用consume命令获取一个队列中的消息。

2.Session Layer，主要负责将客户端的命令发送给服务器，在将服务器端的应答返回给客户端，主要为客户端与服务器之间通信提供可靠性、同步机制和错误处理。

3.Transport Layer，主要传输二进制数据流，提供帧的处理、信道复用、错误检测和数据表示。

RabbitMQ使用场景

学习RabbitMQ的使用场景，来自官方教程：https://www.rabbitmq.com/getstarted.html

场景1：单发送单接收

使用场景：简单的发送与接收，没有特别的处理。

![171507104838529.png][]

Producer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    
    public class Send {
        
      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);
        String message = "Hello World!";
        channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
        
        channel.close();
        connection.close();
      }
    }

![复制代码][copycode.gif]

Consumer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.QueueingConsumer;
    
    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");
    	    
    	
    //	    channel.basicConsume(QUEUE_NAME, true, consumer);
    	    
    	    while (true) {
    	    	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(" Consumer have received '" + message + "'");  
                    }  
                };  
                channel.basicConsume(QUEUE_NAME, true, consumer);  
    	    }
    	  }
    }

![复制代码][copycode.gif]

注1：queueDeclare第一个参数表示队列名称、第二个参数为是否持久化（true表示是，队列将在服务器重启时生存）、第三个参数为是否是独占队列（创建者可以使用的私有队列，断开后自动删除）、第四个参数为当所有消费者客户端长时间连接断开时是否自动删除队列、第五个参数为队列的其他参数

注2：basicPublish第一个参数为交换机名称、第二个参数为队列映射的路由key、第三个参数为消息的其他属性、第四个参数为发送信息的主体

场景2：单发送多接收

使用场景：一个发送端，多个接收端，如分布式的任务派发。为了保证消息发送的可靠性，不丢失消息，使消息持久化了。同时为了防止接收端在处理消息时down掉，只有在消息处理完成后才发送ack消息。

![171513043112193.png][]

Producer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.MessageProperties;
    
    public class NewTask {
      private static final String TASK_QUEUE_NAME="task_queue";
        public static void main(String[] args) throws IOException, TimeoutException {
            ConnectionFactory factory=new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection=factory.newConnection();
            Channel channel=connection.createChannel();
            channel.queueDeclare(TASK_QUEUE_NAME,true,false,false,null);
            //分发信息
            for (int i=0;i<10;i++){
                String message="Hello RabbitMQ"+i;
                channel.basicPublish("",TASK_QUEUE_NAME,
                        MessageProperties.PERSISTENT_TEXT_PLAIN,message.getBytes());
                System.out.println("NewTask send '"+message+"'");
            }
            channel.close();
            connection.close();
        }
    }

![复制代码][copycode.gif]

发送端和场景1不同点：

1、使用“task\_queue”声明了另一个Queue，因为RabbitMQ不容许声明2个相同名称、配置不同的Queue

2、使"task\_queue"的Queue的durable的属性为true，即使消息队列durable

3、使用MessageProperties.PERSISTENT\_TEXT\_PLAIN使消息durable

When RabbitMQ quits or crashes it will forget the queues and messages unless you tell it not to. Two things are required to make sure that messages aren't lost: we need to mark both the queue and messages as durable.

Consumer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.QueueingConsumer;
      
    public class Worker {
    
        //设置和send 相同，
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            //打开connection 和 channel
            Connection connection = factory.newConnection();
            final Channel channel = connection.createChannel();
            /**
             * 声明要消费的queue。可能消费都先被执行，在消费消息之前要确保queue存在。
             * RabbitmqConfigure.DURABLE 设置消息持久化，设置了持久化也不能保证持久化
             *
             */
            channel.queueDeclare("task_queue", true, false, false, null);
            //公平调度：告诉RabbitMQ，再同一时刻，不要发送超过1条消息给一个工作者（worker），直到它已经处理了上一条消息并且作出了响应。
            channel.basicQos(1);
            System.out.println(" [*] Waiting for messages.");
            /**
             * consumer 接收消息回调方法，DefaultConsumer提供一个方法可以缓存发送的消息，直到消息被消费。
             */
            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("Received '" + message + "'");
                    try {
                        doWork(message);
                    }catch(Exception e){
                        e.printStackTrace();
                    } finally {
                        System.out.println("basicAck  Done");
                        //手动确认
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            };
            //设置自动确认为false
            channel.basicConsume("task_queue",  false, consumer);
        }
        private static void doWork(String task) throws InterruptedException {
            for (char ch: task.toCharArray()) {
                if (ch == '.') Thread.sleep(1000);
            }
        }
    }

![复制代码][copycode.gif]

接收端和场景1不同点：

1、使用“task\_queue”声明消息队列，并使消息队列durable

2、在使用channel.basicConsume接收消息时使autoAck为false，即不自动会发ack，由channel.basicAck()在消息处理完成后发送消息。

3、使用了channel.basicQos(1)保证在接收端一个消息没有处理完时不会接收另一个消息，即接收端发送了ack后才会接收下一个消息。在这种情况下发送端会尝试把消息发送给下一个not busy的接收端。

注意点：

1）It's a common mistake to miss the basicAck. It's an easy error, but the consequences are serious. Messages will be redelivered when your client quits (which may look like random redelivery), but RabbitMQ will eat more and more memory as it won't be able to release any unacked messages.

2）Note on message persistence

Marking messages as persistent doesn't fully guarantee that a message won't be lost. Although it tells RabbitMQ to save the message to disk, there is still a short time window when RabbitMQ has accepted a message and hasn't saved it yet. Also, RabbitMQ doesn't do fsync(2) for every message -- it may be just saved to cache and not really written to the disk. The persistence guarantees aren't strong, but it's more than enough for our simple task queue. If you need a stronger guarantee you can wrap the publishing code in atransaction.

3）Note about queue size

If all the workers are busy, your queue can fill up. You will want to keep an eye on that, and maybe add more workers, or have some other strategy.

4）RabbitMQ allows you to set Time To Live for both messages and queues. https://www.rabbitmq.com/ttl.html

场景3：Publish/Subscribe

使用场景：发布、订阅模式，发送端发送广播消息，多个接收端接收。

![171657207955618.png][]

Producer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    
    public class EmitLog {
    
      private static final String EXCHANGE_NAME = "logs";
    
      public static void main(String[] argv) throws Exception {
    
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
    
        String message = getMessage(argv);
    
        channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
        System.out.println(" [x] Sent '" + message + "'");
    
        channel.close();
        connection.close();
      }
      
      private static String getMessage(String[] strings){
        if (strings.length < 1)
                return "info: 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();
      }
    }

![复制代码][copycode.gif]

发送端：

发送消息到一个名为“logs”的exchange上，使用“fanout”方式发送，即广播消息，不需要使用queue，发送端不需要关心谁接收。

Consumer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class ReceiveLogs {
    
      private static final String EXCHANGE_NAME = "logs";
    
      public static void main(String[] argv) throws Exception {
    
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
        String queueName = channel.queueDeclare().getQueue();
        channel.queueBind(queueName, EXCHANGE_NAME, "");
        
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
        QueueingConsumer consumer = new QueueingConsumer(channel);
        channel.basicConsume(queueName, true, consumer);
    
        while (true) {
          QueueingConsumer.Delivery delivery = consumer.nextDelivery();
          String message = new String(delivery.getBody());
    
          System.out.println(" [x] Received '" + message + "'");   
        }
      }
    }

![复制代码][copycode.gif]

接收端：

1、声明名为“logs”的exchange的，方式为"fanout"，和发送端一样。

2、channel.queueDeclare().getQueue();该语句得到一个随机名称的Queue，该queue的类型为non-durable、exclusive、auto-delete的，将该queue绑定到上面的exchange上接收消息。

3、注意binding queue的时候，channel.queueBind()的第三个参数Routing key为空，即所有的消息都接收。如果这个值不为空，在exchange type为“fanout”方式下该值被忽略！

场景4：Routing (按路线发送接收)

使用场景：发送端按routing key发送消息，不同的接收端按不同的routing key接收消息。

![171709256231158.png][]

Producer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    
    public class EmitLogDirect {
    
      private static final String EXCHANGE_NAME = "direct_logs";
    
      public static void main(String[] argv) throws Exception {
    
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
    
        String severity = getSeverity(argv);
        String message = getMessage(argv);
    
        channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
        System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
    
        channel.close();
        connection.close();
      }
      
      private static String getSeverity(String[] strings){
        if (strings.length < 1)
                return "info";
        return strings[0];
      }
    
      private static String getMessage(String[] strings){ 
        if (strings.length < 2)
                return "Hello World!";
        return joinStrings(strings, " ", 1);
      }
      
      private static String joinStrings(String[] strings, String delimiter, int startIndex) {
        int length = strings.length;
        if (length == 0 ) return "";
        if (length < startIndex ) return "";
        StringBuilder words = new StringBuilder(strings[startIndex]);
        for (int i = startIndex + 1; i < length; i++) {
            words.append(delimiter).append(strings[i]);
        }
        return words.toString();
      }
    }

![复制代码][copycode.gif]

发送端和场景3的区别：

1、exchange的type为direct

2、发送消息的时候加入了routing key

Consumer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class ReceiveLogsDirect {
    
      private static final String EXCHANGE_NAME = "direct_logs";
    
      public static void main(String[] argv) throws Exception {
    
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
    
        channel.exchangeDeclare(EXCHANGE_NAME, "direct");
        String queueName = channel.queueDeclare().getQueue();
        
        if (argv.length < 1){
          System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
          System.exit(1);
        }
        
        for(String severity : argv){    
          channel.queueBind(queueName, EXCHANGE_NAME, severity);
        }
        
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
        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();
    
          System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");   
        }
      }
    }

![复制代码][copycode.gif]

接收端和场景3的区别：

在绑定queue和exchange的时候使用了routing key，即从该exchange上只接收routing key指定的消息。

场景5：Topics (按topic发送接收)

使用场景：发送端不只按固定的routing key发送消息，而是按字符串“匹配”发送，接收端同样如此。

![171739210926049.png][]

Producer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    
    public class EmitLogTopic {
    
      private static final String EXCHANGE_NAME = "topic_logs";
    
      public static void main(String[] argv) {
        Connection connection = null;
        Channel channel = null;
        try {
          ConnectionFactory factory = new ConnectionFactory();
          factory.setHost("localhost");
      
          connection = factory.newConnection();
          channel = connection.createChannel();
    
          channel.exchangeDeclare(EXCHANGE_NAME, "topic");
    
          String routingKey = getRouting(argv);
          String message = getMessage(argv);
    
          channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
          System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
    
        }
        catch  (Exception e) {
          e.printStackTrace();
        }
        finally {
          if (connection != null) {
            try {
              connection.close();
            }
            catch (Exception ignore) {}
          }
        }
      }
      
      private static String getRouting(String[] strings){
        if (strings.length < 1)
                return "anonymous.info";
        return strings[0];
      }
    
      private static String getMessage(String[] strings){ 
        if (strings.length < 2)
                return "Hello World!";
        return joinStrings(strings, " ", 1);
      }
      
      private static String joinStrings(String[] strings, String delimiter, int startIndex) {
        int length = strings.length;
        if (length == 0 ) return "";
        if (length < startIndex ) return "";
        StringBuilder words = new StringBuilder(strings[startIndex]);
        for (int i = startIndex + 1; i < length; i++) {
            words.append(delimiter).append(strings[i]);
        }
        return words.toString();
      }
    }

![复制代码][copycode.gif]

发送端和场景4的区别：

1、exchange的type为topic

2、发送消息的routing key不是固定的单词，而是匹配字符串，如"\*.lu.\#"，\*匹配一个单词，\#匹配0个或多个单词。

Consumer：

![复制代码][copycode.gif]

import com.rabbitmq.client.ConnectionFactory;
    import com.rabbitmq.client.Connection;
    import com.rabbitmq.client.Channel;
    import com.rabbitmq.client.QueueingConsumer;
    
    public class ReceiveLogsTopic {
    
      private static final String EXCHANGE_NAME = "topic_logs";
    
      public static void main(String[] argv) {
        Connection connection = null;
        Channel channel = null;
        try {
          ConnectionFactory factory = new ConnectionFactory();
          factory.setHost("localhost");
      
          connection = factory.newConnection();
          channel = connection.createChannel();
    
          channel.exchangeDeclare(EXCHANGE_NAME, "topic");
          String queueName = channel.queueDeclare().getQueue();
     
          if (argv.length < 1){
            System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
            System.exit(1);
          }
        
          for(String bindingKey : argv){    
            channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
          }
        
          System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
    
          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();
    
            System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");   
          }
        }
        catch  (Exception e) {
          e.printStackTrace();
        }
        finally {
          if (connection != null) {
            try {
              connection.close();
            }
            catch (Exception ignore) {}
          }
        }
      }
    }

![复制代码][copycode.gif]

接收端和场景4的区别：

1、exchange的type为topic

2、接收消息的routing key不是固定的单词，而是匹配字符串。

[171507104838529.png]: /images/20220608/6f244d460e9d42e3985285c9528bfbaf.png
[copycode.gif]: /images/20220608/aab4a9c6a124449a8ab67fc519c73c22.png
[171513043112193.png]: /images/20220608/f8074a68193745b8a7b31e55200cff86.png
[171657207955618.png]: /images/20220608/76fc69f455b84d958e5e847794ecbc7d.png
[171709256231158.png]: /images/20220608/778107bc000a4815b127617a5108ceae.png
[171739210926049.png]: /images/20220608/2fc8dc574f2e4ba9823e487518d0cae6.png