Work Queues

  1. 工作队列(任务队列)主要思想是避免立即执行资源密集型任务，而不得不等待它完成。
  2. 我们将任务封装成消息发送到队列中，在后台的工作线程(消费者)弹出任务并执行。
  3. 当有多个工作线程一起处理任务时，一个消息只能被消费一次，默认轮询策略分发机制。

RabbitMQ连接工具类

  1. public class RabbitMqUtils {
  2.     //获取连接
  3.     public static Connection getConnection() {
  4.         //创建一个连接工厂
  5.         ConnectionFactory factory = new ConnectionFactory();
  6.         //设置工厂IP 连接RabbitMQ的队列
  7.         factory.setHost("47.172.193.131");
  8.         factory.setPort(5672);
  9.         factory.setUsername("admin");
  10.         factory.setPassword("123");
  11.         factory.setVirtualHost("/");
  12.         Connection connection = null;
  13.         try {
  14.             connection = factory.newConnection();
  15.         } catch (IOException | TimeoutException e) {
  16.             e.printStackTrace();
  17.         }
  18.         return connection;
  19.     }
  20.     //得到一个信道
  21.     public static Channel getChannel() {
  22.         Channel channel = null;
  23.         try {
  24.             channel = getConnection().createChannel();
  25.         } catch (IOException e) {
  26.             e.printStackTrace();
  27.         }
  28.         return channel;
  29.     }
  30.     //资源关闭
  31.     public static void close(Connection connection, Channel channel) {
  32.         if (channel != null) {
  33.             try {
  34.                 channel.close();
  35.             } catch (IOException | TimeoutException e) {
  36.                 e.printStackTrace();
  37.             }
  38.         }
  39.         if (connection != null) {
  40.             try {
  41.                 connection.close();
  42.             } catch (IOException e) {
  43.                 e.printStackTrace();
  44.             }
  45.         }
  46.     }
  47. }

消息轮询消费

  1. 工作线程1                                         工作线程2
  2. task0                                            task1
  3. task2                                            task3
  4. task4                                              task5

消息应答

  1. RabbitMQ一但向消费者投递一条消息便立即标记为删除，如果消费者在处理消息的过程中宕机了，将丢失正在
  2. 处理的消息。为了保证消息在发送过程中不丢失，RabbitMQ引入了应答机制，在消费者处理消息完毕后，
  3. 告诉RabbitMQ可以删除消息了。
  4. 自动应答:
  5.     不在乎消费者对消息处理是否成功，都会告诉队列删除消息。如果处理消息失败，TCP连接断开，
  6.   RabbitMQ将实现自动补偿，重新将消息发送到其他消费者。
  7. 手动应答:
  8.     消费者处理完业务逻辑，手动返回Ack（通知）告诉队列处理完了，队列进而删除消息。
  9. //手动应答 参数1:消息的唯一标识    参数2:是否批量应答
  10. channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);

消息持久化

  1. RabbitMQ的消息和队列默认是在内存中的，如果Rabbit Server宕机会导致消息丢失，
  2. 我们可以通过标记消息和队列为持久化避免这个问题。
  3. #创建队列第二个参数为是否创建持久化队列
  4. queueDeclare(QUEUE_NAME, true, false, false, null);
  5. #MessageProperties.PERSISTENT_TEXT_PLAIN:标记消息为持久化
  6. #但不能完全保证消息不会丢失，如果消息在内存写入磁盘的那一瞬间宕机依然会导致消息丢失。
  7. channel.basicPublish("", QUEUE_NAME, MessageProperties.PERSISTENT_TEXT_PLAIN,
  8.                                                     message.getBytes(StandardCharsets.UTF_8));

不公平分发

  1. RabbitMQ默认采用轮询分发，如果消费者处理消息能力不同，会导致处理速度快的消费者和处理速度慢的
  2. 消费者得到同样数量的消息，这样会使处理能力快的消费者大部分时间都处于空闲状态，造成资源浪费。
  3. #消费方设置分发策略为不公平分发
  4. channel.basicQos(1)

预取值

  1. 可以手动设置某个消费者单次接收多少条消息进行处理，设置信道的容量。
  2. int prefetchCount = 5;#设置预取值为5
  3. channel.basicQos(prefetchCount);

发布确认

  1. 在消息持久化的过程中仍然存在极小概率出现缓存写入磁盘宕机导致数据丢失的可能，需要采用发布确认保证
  2. 消息不丢失。发布确认就是在消息从内存写入到磁盘完毕后，需要响应给消息提供者持久化完成，才算持久化成功。
  3. #发布确认默认是不开启的，需要调用channel的confirmSelect()方法
  4. channel.confirmSelect();
  5. #1.单个确认发布
  6. 同步确认发布，发布一个消息后只有在它被确认发布后，后续发布才能进行，如果在指定的时间内没有返回确认
  7. 消息就会抛异常。
  8. 缺点:发布速度比较慢，因为消息的确认过程是阻塞的，一般适用于每秒不超过数百条发布消息吞吐量的小型应用程序。
  9. channel.confirmSelect();#开启单个发布确认
  10. channel.waitForConfirms();#确认消息发布，返回true为发布成功
  11. #2.批量确认发布
  12. 先发布一批消息然后一起确认，相比于单个发布确认提高了吞吐量，缺点就是如果发生故障导致发布出现问题，
  13. 无法定位到出现问题的消息，我们必须将整个批处理保存在内存中，以记录重要的信息而后重新发布，同时也是阻塞的。
  14. 在单个确认发布的基础上指定一个量，达到这个量就进行一次确认发布。
  15. #3.异步确认发布
  16. 可靠性和性能都能得到极大的保证，需要一个Map标记消息的key和value，利用回调函数来达到消息可靠性传递，
  17. ack确认应答回调和nack未确认应答回调函数，非阻塞型。
  18. //消息确认成功回调函数 ack
  19. ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
  20.     //消息成功回调函数
  21.     System.out.println("消息确认成功" + deliveryTag);
  22. };
  23. //消息确认失败回调函数 nack
  24. ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
  25.     System.out.println("消息确认失败" + deliveryTag);
  26. };
  27. channel.addConfirmListener(ackCallback, nackCallback);
  28. #处理异步未确认消息
  29. 最好就是把未确认的消息放到一个基于内存的能被发布线程访问的队列，例如用ConcurrentSkipListMap可以在
  30. confirm callbacks与发布线程之间进行消息的传递。
  32. [单次确认发布耗时(1800,2000)毫秒]
  33. [批量确认发布耗时(200,300)毫秒]
  34. [异步确认发布耗时(90,110)毫秒]
  1. //记录所有发布的消息 key:序号 value:消息体
  2. outstandingConfirms.put(channel.getNextPublishSeqNo(), message + i);
  1. /*
  2.  * 线程安全有序的一个哈希表，适用于高并发的情况下
  3.  * 1.轻松的将序号和消息关联
  4.  * 2.可以轻松的批量删除
  5.  * 3.支持高并发
  6.  */
  7. ConcurrentSkipListMap<Long, String> outstandingConfirms =
  8.     new ConcurrentSkipListMap<>();
  9. //消息确认成功回调函数 ack
  10. ConfirmCallback ackCallback = (deliveryTag, multiple) -> {
  11.     //消息成功回调函数 - 删除确认消息编号(剩下的是未被确认的消息)
  12.     if (multiple) {
  13.         //批量删除
  14.         ConcurrentNavigableMap<Long, String> confirmed 
  15.             = outstandingConfirms.headMap(deliveryTag);
  16.         confirmed.clear();
  17.     } else {
  18.         outstandingConfirms.remove(deliveryTag);
  19.     }
  20. };
  21. //消息确认失败回调函数 nack
  22. ConfirmCallback nackCallback = (deliveryTag, multiple) -> {
  23.     System.out.println("未确认消息" + outstandingConfirms.get(deliveryTag));
  24. };
  25. channel.addConfirmListener(ackCallback, nackCallback);