RabbitMQ

冷不防 2022-05-31 07:39 86阅读 0赞

RabbitMQ

消息中间件














































名称 图标 简介 适用场景 优点 缺点
RabbitMQ 是AMQP协议领先的一个实现，他实现了代理架构，意味着在发送到客户端之前可以在中央节点上排队，此特性使得RabbitMQ易于使用和部署 路由、负载均衡、消息持久化 几行代码就可以搞定 扩展性差，速度较慢—中央节点增加了延迟，消息封装后也比较大
ZeroMQ 是一个非常轻量级的消息系统，专门为高吞吐量/低延迟的场景开发 金融界 与RabbitMQ相比，ZeroMQ支持许多高级的场景但是必须实现ZeroMQ的框架中的各个快（比如Socket和Device等）；非常灵活 手册有80多页，如果手写一个分布式系统，一定得阅读它
ActiveMQ 居于两者之间 类似于RabbitMQ，可以部署于代理模式和p2p模式；类似于ZeroMQ，易于实现高级场景,而且只需付出低消耗 被视为消息中间件的“瑞士军刀” 因为吸收两者的优点，暂时略缺点
Apollo 下一代的ActiveMQ 越来越多领域喽 越来越好喽 这个暂时略

RabbitMQ简单学习—从京淘（电商网）项目入手

  • 引入

    • 当前京淘的架构性能提升点

      1. NGINX高并发
      2. Redis内存缓存数据库（非结构数据）
      3. amoeba提升数据最后关卡的性能

    • 超负荷的请求，以上三个技术无法处理

      • 当请求来到时，如果并发量太大，就让请求排成队列
    • 基于erlang语言

  • 消息队列分类

    • simple简单队列【先后顺序】
    • work工作模式【资源竞争】—红包
    • publish/subscribe发布订阅【共享资源】：引入交换机—邮件的群发、群聊天、广播
    • 路由模式：消息的生产者发送给交换机，通过路由判断key值发送到相应的队列—error通知
    • topic主题模式（路由模式的一种）：通过表达式进行判断—*代表多个单词，#号代表一个单词

  • 注意：别名

    • publish：fanout
    • routing：direct
    • topic:topic

使用

  • 依赖

    #

    1. <dependency>
    2.     <groupId>com.rabbitmq</groupId>
    3.     <artifactId>amqp-client</artifactId>
    4.     <version>3.5.1</version>
    5. </dependency>

  • 流程

    1. 创建连接工厂
    2. 从连接工厂获取connection
    3. 从连接获取channel
    4. 从channel获取绑定的queue
    5. 生产者生产消息放入队列
    6. 释放资源

RabbitMQ的工作原理

  • 单发送，单接收

    • 使用场景：简单的发送与接收，没有设么特别的处理
    • 0QPW6QC.png

    • 示例【生产者】

      #

      1. public class Send {
      2.   private final static String QUEUE_NAME = "hello";//队列的名称
      4.   public static void main(String[] argv) throws Exception {
      5.     // 获取连接工厂
      6.     ConnectionFactory factory = new ConnectionFactory();
      7.     // 设置主机IP
      8.     factory.setHost("localhost");
      9.     // 获取连接
      10.     Connection connection = factory.newConnection();
      11.     // 创建通道
      12.     Channel channel = connection.createChannel();
      13.     // 通道找到队列
      14.     channel.queueDeclare(QUEUE_NAME, false, false, false, null);
      15.     String message = "Hello World!";
      16.     // 发送消息给队列
      17.     channel.basicPublish("", QUEUE_NAME, null, message.getBytes());
      18.     System.out.println(" [x] Sent '" + message + "'");
      19.     // 关闭连接
      20.     channel.close();
      21.     connection.close();
      22.   }
      23. }

    • 示例【消费者】

      #

      1. public class Recv {
      2.     // 队列名称
      3.     private final static String QUEUE_NAME = "hello";
      5.     public static void main(String[] argv) throws Exception {
      6.     // 获得连接工厂
      7.     ConnectionFactory factory = new ConnectionFactory();
      8.     // 设置主机IP
      9.     factory.setHost("localhost");
      10.     // 获得连接
      11.     Connection connection = factory.newConnection();
      12.     // 创建通道
      13.     Channel channel = connection.createChannel();
      14.     // 通道连接队列
      15.     channel.queueDeclare(QUEUE_NAME, false, false, false, null);
      16.     System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
      17.     // 接收队列
      18.     QueueingConsumer consumer = new QueueingConsumer(channel);
      19.     // 执行
      20.     channel.basicConsume(QUEUE_NAME, true, consumer);
      21.     // 遍历队列消息
      22.     while (true) {
      23.     // 传送队列信息
      24.       QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      25.       String message = new String(delivery.getBody());
      26.       System.out.println(" [x] Received '" + message + "'");
      27.     }
      28.   }
      29. }

  • 单发送多接收

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

    • 示例【生产者】

      #

      1. public class NewTask {
      2.   // 队列的名称
      3.   private static final String TASK_QUEUE_NAME = "task_queue";
      5.   public static void main(String[] argv) throws Exception {
      6.     // 连接工厂
      7.     ConnectionFactory factory = new ConnectionFactory();
      8.     // 设置主机IP
      9.     factory.setHost("localhost");
      10.     // 获取连接
      11.     Connection connection = factory.newConnection();
      12.     // 创建通道
      13.     Channel channel = connection.createChannel();
      15.     channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
      17.     String message = getMessage(argv);
      18.     // PERSISTENT_TEXT_PLAIN:消息的持久化
      19.     channel.basicPublish("", TASK_QUEUE_NAME, 
      20.                 MessageProperties.PERSISTENT_TEXT_PLAIN,
      21.                 message.getBytes());
      22.     System.out.println(" [x] Sent '" + message + "'");
      24.     channel.close();
      25.     connection.close();
      26.   }
      28.   private static String getMessage(String[] strings){
      29.     if (strings.length < 1)
      30.       return "Hello World!";
      31.     return joinStrings(strings, " ");
      32.   }  
      34.   private static String joinStrings(String[] strings, String delimiter) {
      35.     int length = strings.length;
      36.     if (length == 0) return "";
      37.     StringBuilder words = new StringBuilder(strings[0]);
      38.     for (int i = 1; i < length; i++) {
      39.       words.append(delimiter).append(strings[i]);
      40.     }
      41.     return words.toString();
      42.   }
      43. }

    • 示例【消费者】

      #

      1. public class Worker {
      3.   private static final String TASK_QUEUE_NAME = "task_queue";
      5.   public static void main(String[] argv) throws Exception {
      7.     ConnectionFactory factory = new ConnectionFactory();
      8.     factory.setHost("localhost");
      9.     Connection connection = factory.newConnection();
      10.     Channel channel = connection.createChannel();
      12.     channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
      13.     System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
      14.     // 使用了channel.basicQos(1)保证在接收端一个消息没有处理完时不会接收另一个消息，即接收端发送了ack后才会接收下一个消息。在这种情况下发送端会尝试把消息发送给下一个not busy的接收端
      15.     channel.basicQos(1);
      17.     QueueingConsumer consumer = new QueueingConsumer(channel);
      18.     channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
      20.     while (true) {
      21.       QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      22.       String message = new String(delivery.getBody());
      24.       System.out.println(" [x] Received '" + message + "'");
      25.       doWork(message);
      26.       System.out.println(" [x] Done");
      28.       channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
      29.     }         
      30.   }
      32.   private static void doWork(String task) throws InterruptedException {
      33.     for (char ch: task.toCharArray()) {
      34.       if (ch == '.') Thread.sleep(1000);
      35.     }
      36.   }
      37. }

  • Publish/Subscribe

    • 使用场景：发布、订阅模式，发送端发送广播消息，多个接收端接收
    • 1jRyhLo.png

    • 示例【生产者】

      #

      1. public class EmitLog {
      3.   private static final String EXCHANGE_NAME = "logs";
      5.   public static void main(String[] argv) throws Exception {
      7.     ConnectionFactory factory = new ConnectionFactory();
      8.     factory.setHost("localhost");
      9.     Connection connection = factory.newConnection();
      10.     Channel channel = connection.createChannel();
      12.     channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
      14.     String message = getMessage(argv);
      16.     channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes());
      17.     System.out.println(" [x] Sent '" + message + "'");
      19.     channel.close();
      20.     connection.close();
      21.   }
      23.   private static String getMessage(String[] strings){
      24.     if (strings.length < 1)
      25.             return "info: Hello World!";
      26.     return joinStrings(strings, " ");
      27.   }
      29.   private static String joinStrings(String[] strings, String delimiter) {
      30.     int length = strings.length;
      31.     if (length == 0) return "";
      32.     StringBuilder words = new StringBuilder(strings[0]);
      33.     for (int i = 1; i < length; i++) {
      34.         words.append(delimiter).append(strings[i]);
      35.     }
      36.     return words.toString();
      37.   }
      38. }

    • 示例【消费者】

      #

      1. public class ReceiveLogs {
      3.   private static final String EXCHANGE_NAME = "logs";
      5.   public static void main(String[] argv) throws Exception {
      7.     ConnectionFactory factory = new ConnectionFactory();
      8.     factory.setHost("localhost");
      9.     Connection connection = factory.newConnection();
      10.     Channel channel = connection.createChannel();
      12.     channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
      13.     String queueName = channel.queueDeclare().getQueue();
      14.     channel.queueBind(queueName, EXCHANGE_NAME, "");
      16.     System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
      18.     QueueingConsumer consumer = new QueueingConsumer(channel);
      19.     channel.basicConsume(queueName, true, consumer);
      21.     while (true) {
      22.       QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      23.       String message = new String(delivery.getBody());
      25.       System.out.println(" [x] Received '" + message + "'");   
      26.     }
      27.   }
      28. }

  • Routing (按路线发送接收)

    • 使用场景：发送端按routing key发送消息，不同的接收端按不同的routing key接收消息
    • fzDXMDh.png

    • 示例【生产者】

      #

      1. public class EmitLogDirect {
      3.   private static final String EXCHANGE_NAME = "direct_logs";
      5.   public static void main(String[] argv) throws Exception {
      7.     ConnectionFactory factory = new ConnectionFactory();
      8.     factory.setHost("localhost");
      9.     Connection connection = factory.newConnection();
      10.     Channel channel = connection.createChannel();
      12.     channel.exchangeDeclare(EXCHANGE_NAME, "direct");
      14.     String severity = getSeverity(argv);
      15.     String message = getMessage(argv);
      17.     channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes());
      18.     System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
      20.     channel.close();
      21.     connection.close();
      22.   }
      24.   private static String getSeverity(String[] strings){
      25.     if (strings.length < 1)
      26.             return "info";
      27.     return strings[0];
      28.   }
      30.   private static String getMessage(String[] strings){ 
      31.     if (strings.length < 2)
      32.             return "Hello World!";
      33.     return joinStrings(strings, " ", 1);
      34.   }
      36.   private static String joinStrings(String[] strings, String delimiter, int startIndex) {
      37.     int length = strings.length;
      38.     if (length == 0 ) return "";
      39.     if (length < startIndex ) return "";
      40.     StringBuilder words = new StringBuilder(strings[startIndex]);
      41.     for (int i = startIndex + 1; i < length; i++) {
      42.         words.append(delimiter).append(strings[i]);
      43.     }
      44.     return words.toString();
      45.   }
      46. }

    • 示例【消费者】

      #

      1. public class ReceiveLogsDirect {
      3.   private static final String EXCHANGE_NAME = "direct_logs";
      5.   public static void main(String[] argv) throws Exception {
      7.     ConnectionFactory factory = new ConnectionFactory();
      8.     factory.setHost("localhost");
      9.     Connection connection = factory.newConnection();
      10.     Channel channel = connection.createChannel();
      12.     channel.exchangeDeclare(EXCHANGE_NAME, "direct");
      13.     String queueName = channel.queueDeclare().getQueue();
      15.     if (argv.length < 1){
      16.       System.err.println("Usage: ReceiveLogsDirect [info] [warning] [error]");
      17.       System.exit(1);
      18.     }
      20.     for(String severity : argv){    
      21.       channel.queueBind(queueName, EXCHANGE_NAME, severity);
      22.     }
      24.     System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
      26.     QueueingConsumer consumer = new QueueingConsumer(channel);
      27.     channel.basicConsume(queueName, true, consumer);
      29.     while (true) {
      30.       QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      31.       String message = new String(delivery.getBody());
      32.       String routingKey = delivery.getEnvelope().getRoutingKey();
      34.       System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");   
      35.     }
      36.   }
      37. }

  • Topics (按topic发送接收)

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

    • 示例【生产者】

      #

      1. public class EmitLogTopic {
      3.   private static final String EXCHANGE_NAME = "topic_logs";
      5.   public static void main(String[] argv) {
      6.     Connection connection = null;
      7.     Channel channel = null;
      8.     try {
      9.       ConnectionFactory factory = new ConnectionFactory();
      10.       factory.setHost("localhost");
      12.       connection = factory.newConnection();
      13.       channel = connection.createChannel();
      15.       channel.exchangeDeclare(EXCHANGE_NAME, "topic");
      17.       String routingKey = getRouting(argv);
      18.       String message = getMessage(argv);
      20.       channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes());
      21.       System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
      23.     }
      24.     catch  (Exception e) {
      25.       e.printStackTrace();
      26.     }
      27.     finally {
      28.       if (connection != null) {
      29.         try {
      30.           connection.close();
      31.         }
      32.         catch (Exception ignore) {}
      33.       }
      34.     }
      35.   }
      37.   private static String getRouting(String[] strings){
      38.     if (strings.length < 1)
      39.             return "anonymous.info";
      40.     return strings[0];
      41.   }
      43.   private static String getMessage(String[] strings){ 
      44.     if (strings.length < 2)
      45.             return "Hello World!";
      46.     return joinStrings(strings, " ", 1);
      47.   }
      49.   private static String joinStrings(String[] strings, String delimiter, int startIndex) {
      50.     int length = strings.length;
      51.     if (length == 0 ) return "";
      52.     if (length < startIndex ) return "";
      53.     StringBuilder words = new StringBuilder(strings[startIndex]);
      54.     for (int i = startIndex + 1; i < length; i++) {
      55.         words.append(delimiter).append(strings[i]);
      56.     }
      57.     return words.toString();
      58.   }
      59. }

    • 示例【消费者】

      #

      1. public class ReceiveLogsTopic {
      3.   private static final String EXCHANGE_NAME = "topic_logs";
      5.   public static void main(String[] argv) {
      6.     Connection connection = null;
      7.     Channel channel = null;
      8.     try {
      9.       ConnectionFactory factory = new ConnectionFactory();
      10.       factory.setHost("localhost");
      12.       connection = factory.newConnection();
      13.       channel = connection.createChannel();
      15.       channel.exchangeDeclare(EXCHANGE_NAME, "topic");
      16.       String queueName = channel.queueDeclare().getQueue();
      18.       if (argv.length < 1){
      19.         System.err.println("Usage: ReceiveLogsTopic [binding_key]...");
      20.         System.exit(1);
      21.       }
      23.       for(String bindingKey : argv){    
      24.         channel.queueBind(queueName, EXCHANGE_NAME, bindingKey);
      25.       }
      27.       System.out.println(" [*] Waiting for messages. To exit press CTRL+C");
      29.       QueueingConsumer consumer = new QueueingConsumer(channel);
      30.       channel.basicConsume(queueName, true, consumer);
      32.       while (true) {
      33.         QueueingConsumer.Delivery delivery = consumer.nextDelivery();
      34.         String message = new String(delivery.getBody());
      35.         String routingKey = delivery.getEnvelope().getRoutingKey();
      37.         System.out.println(" [x] Received '" + routingKey + "':'" + message + "'");   
      38.       }
      39.     }
      40.     catch  (Exception e) {
      41.       e.printStackTrace();
      42.     }
      43.     finally {
      44.       if (connection != null) {
      45.         try {
      46.           connection.close();
      47.         }
      48.         catch (Exception ignore) {}
      49.       }
      50.     }
      51.   }
      52. }

秒杀

业务场景分析

  • 并发量很高的时间段—抢商品

  • 队列中的消息可以是什么

    1. 电话号码
    2. username
    3. ticket
    4. ……

  • 做法

    • 调用SSO查询用户信息，把前n个消息获取到，后面的放入rabbitmq的垃圾桶
    • 更高的并发可以考虑分布式的队列

  • 文件位置

    • 生产者：后台
    • 消费者：前台

  • 秒杀设计

    • KoWfMTK.png
  • 未完待续。。。

注：参考文章

  • RabbitMQ的几种典型使用场景：https://www.rabbitmq.com/getstarted.html

发表评论

表情：
评论列表 （有 0 条评论，86人围观）

还没有评论，来说两句吧...

相关阅读

    相关 RabbitMQ(一)：初始RabbitMQ

    1. 消息队列的作用 消息队列中间件是分布式系统中重要的组件，主要解决应用解耦，异步消息，流量削锋等问题，实现高性能，高可用，可伸缩和最终一致性架构。 2. Rabb

    短命女短命女/ 0 赞/ 120 阅读