RabbitMQ深度探索：创建消息队列

快速入门：实现 RabbitMQ 简单队列：

• 在 RabbitMQ 平台创建 Virtual Hosts 和一个队列
  
• /boyaVirtualHosts
  
  
  • 订单队列
    
  • 付出队列
    
  
  
• 导入依赖：
  
  
  1. <dependency>
  2.   <groupId>com.rabbitmq</groupId>
  3.   <artifactId>amqp-client</artifactId>
  4.   <version>3.6.5 </version>
  5. </dependency>

  复制代码
• 编写毗连类：
  
  
  1. public class RabbitMQConnection {
  2.     /**
  3.      * 获取连接
  4.      */
  5.     public static Connection getConnection() throws IOException, TimeoutException {
  6.         // 1.创建连接
  7.         ConnectionFactory connectionFactory = new ConnectionFactory();
  9.         // 2.设置连接地址
  10.         connectionFactory.setHost("127.0.0.1");
  12.         // 3.设置端口号
  13.         connectionFactory.setPort(5672);
  15.         // 4.设置账号和密码
  16.         connectionFactory.setUsername("guest");
  17.         connectionFactory.setPassword("guest");
  19.         // 5.设置 VirtualHost
  20.         connectionFactory.setVirtualHost("/boyaVirtualHostsR");
  22.         return connectionFactory.newConnection();
  23.     }
  24. }

  复制代码
• 编写生产者代码：
  
  
  1. public class Producer {
  2.     private static final String QUEUE_NAME = "BoyatopMamber";
  3.     /**
  4.      * 获取连接
  5.      */
  6.     public static void main(String[] args) throws IOException, TimeoutException {
  9.         while (true){
  10.             // 1.创建连接
  11.             Connection connection = RabbitMQConnection.getConnection();
  13.             // 2.设置通道
  14.             Channel channel = connection.createChannel();
  16.             // 3.设置消息
  17.             String msg = "Hello World";
  18.             System.out.println("msg:" + msg);
  19.             channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
  22.             channel.close();
  23.             connection.close();
  24.         }
  25.     }
  26. }

  复制代码
• 编写消费者代码：
  
  
  1. public class Comsumer {
  2.     private static final String QUEUE_NAME = "BoyatopMamber";
  4.     public static void main(String[] args) throws IOException, TimeoutException {
  5.         // 1.创建链接
  6.         Connection connection = RabbitMQConnection.getConnection();
  8.         // 2.设置通道
  9.         Channel channel = connection.createChannel();
  10.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
  11.             @Override
  12.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
  13.                 String msg = new String(body,"UTF-8");
  14.                 System.out.println("消费者获取消息：" + msg);
  15.             }
  16.         };
  18.         // 3.监听队列
  19.         channel.basicConsume(QUEUE_NAME,true,defaultConsumer);
  20.     }
  21. }

  复制代码
• RabbitMQ 怎样包管消息不丢失：
  
  
  • 生产者角色：
    
    
    • 确保生产者角色投递到 MQ 服务器端乐成
      
    • Ack 消息确认机制
      
    • 同步或异步的形式：
      
      
      • Confirms
        
      • 事件消息
        
      
      
    
    
  • 消费者角色：
    
    
    • 在 RabbitMQ 情况下：
      
      
      • 必须要将消息消费乐成之后，才会将消息从 MQ 服务器端移除
        
      
      
    • 在 kafka 中的情况下：
      
      
      • 不管是消费乐成照旧消费失败，该消息都不会立即从 MQ 服务器移除
        
      
      
    
    
  • MQ 服务器端：
    
    
    • 在默认的情况下，都会对队列中的消息长期化，长期化硬盘
      
    
    
  
  
• 利用消息确认机制 + 长期化技术实现：
  
  
  • A 消费者确认收到消息机制
    1. channel.basicConsume(QUEUE_NAME,false,defaultConsumer);

    复制代码
    
    
    • 第二个参数值为 false，代表关闭 RabbitMQ 的主动应答机制，改为手动应答
      
    • 在处置惩罚完消息时，返回应答状态，true 表示为主动应答模式
      1. channel.basicAck(envelope.getDeliveryTag(),false);

      复制代码
    
    
  • B 生产者确认投递消息乐成，利用 Confirm 机制或者事件消息
    
    
    
  • Confirm 机制，同步或异步的形式
    
  
  
• RabbitMQ 默认创建是长期化的形式：
  
  
  
  • 将代码中的 durable 设为 true
    
  • 参数详解：
    
    
    • Durability：是否长期化
      
      
      • durable：长期化
        
      • Transient：不长期化
        
      
      
    • Auto delete：是否主动删除
      
      
      • 当最后一个消费者断开毗连之后队列是否主动被删除
        
      • 可以通过 RabbitMQ Management 查看某个队列的消费者数量
        
      • 当 consumers = 0 时，队列就会主动删除
        
      
      
    
    
  
  
• 利用 RabbitMQ 事件：
  1. //设置事务
  2. channel.txSelect();
  3. channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
  4. channel.txCommit();

  复制代码
  
  
  • 生产者：
    2. public class producer {
    3.     private static final String QUEUE_NAME = "BoyatopMamber";
    5.     public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
    6.         // 1.创建新的连接
    7.         Connection connection = RabbitMQConnection.getConnection();
    9.         // 2.设置 channel
    10.         Channel channel = connection.createChannel();
    12.         // 3.发送消息
    13.         String msg = "Hello my Bro";
    14.         channel.confirmSelect();
    16.         channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
    17.         boolean result = channel.waitForConfirms();
    19.         if(result){
    20.             System.out.println("消息投递成功");
    21.         }else {
    22.             System.out.println("消息投递失败");
    23.         }
    25.         // 4.关闭资源
    26.         channel.close();
    27.         connection.close();
    28.     }
    29. }

    复制代码
  • 消费者：
    1. public class Consumer {
    2.     private static final String QUEUE_ANME = "BoyatopMamber";
    4.     public static void main(String[] args) throws IOException, TimeoutException {
    5.         // 1.创建连接
    6.         Connection connection = RabbitMQConnection.getConnection();
    8.         //2.设置通道
    9.         Channel channel = connection.createChannel();
    10.         DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
    11.             @Override
    12.             public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
    13.                 String msg = new String(body,"UTF-8");
    14.                 System.out.println("消费者获取消息:" + msg);
    16.                 //消费者完成 消费该消息
    17.                 channel.basicAck(envelope.getDeliveryTag(),false);
    18.             }
    19.         };
    21.         // 3.监听队列
    22.         channel.basicConsume(QUEUE_ANME,false,defaultConsumer);
    23.     }
    24. }

    复制代码
  
  

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。