消息队列_Rabbitmq

﹏ヽ暗。殇╰゛Y 2021-10-06 03:40 711阅读 0赞

### 消息队列\_Rabbitmq ###

*  简介
 *  场景简介
 *   *  基础获取连接
     *  手动应答
     *  一对一 Work Queues
     *  一对多 Publish/Subscribe
     *  指定发送路由 Routing
     *  通配符匹配 Topic

# 简介 #

RabbitMQ是实现了高级消息队列协议（AMQP）的开源消息代理软件（亦称面向消息的中间件）。RabbitMQ服务器是用Erlang语言编写的，而群集和故障转移是构建在开放电信平台框架上的。所有主要的编程语言均有与代理接口通讯的客户端库。

官方：www.rabbitmq.com

# 场景简介 #

## 基础获取连接 ##

//通过连接工厂创建新的连接和mq建立连接
    ConnectionFactory connectionFactory = new ConnectionFactory();
    connectionFactory.setHost("127.0.0.1");
    connectionFactory.setPort(5672);//端口
    connectionFactory.setUsername("guest");
    connectionFactory.setPassword("guest");
    //设置虚拟机，一个mq服务可以设置多个虚拟机，每个虚拟机就相当于一个独立的mq
    connectionFactory.setVirtualHost("/");
    
    //建立新连接
    Connection connection = connectionFactory.newConnection();
    //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
    Channel channel = connection.createChannel();

## 手动应答 ##

消费方：  
channel.basicReject(envelope.getDeliveryTag(), true); //重新放入队列  
channel.basicAck(envelope.getDeliveryTag(), false); //手动应答

//实现消费方法
      DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
          /**
           * 当接收到消息后此方法将被调用
           * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
           * @param envelope 信封，通过envelope
           * @param properties 消息属性
           * @param body 消息内容
           * @throws IOException
           */
          @Override
          public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
              //交换机
              String exchange = envelope.getExchange();
              //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
              long deliveryTag = envelope.getDeliveryTag();
              //消息内容
              String message= new String(body,"utf-8");
              System.out.println("receive message:"+message);
              channel.basicReject(deliveryTag,true); //重新放入队列
              channel.basicAck(deliveryTag,false); //手动应答
          }
      };
      //监听队列
      //参数：String queue, boolean autoAck, Consumer callback
      /**
       * 参数明细：
       * 1、queue 队列名称
       * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
       * 3、callback，消费方法，当消费者接收到消息要执行的方法
       */
      channel.basicConsume(QUEUE,false,defaultConsumer);

## 一对一 Work Queues ##

（1）生产方：

（2）消费方：

//监听队列
    //声明队列，如果队列在mq 中没有则要创建
    //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
    /**
     * 参数明细
     * 1、queue 队列名称
     * 2、durable 是否持久化，如果持久化，mq重启后队列还在
     * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
     * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
     * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
     */
    channel.queueDeclare(QUEUE,true,false,false,null);
    
    //实现消费方法
    DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
        /**
         * 当接收到消息后此方法将被调用
         * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
         * @param envelope 信封，通过envelope
         * @param properties 消息属性
         * @param body 消息内容
         * @throws IOException
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            //交换机
            String exchange = envelope.getExchange();
            //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
            long deliveryTag = envelope.getDeliveryTag();
            //消息内容
            String message= new String(body,"utf-8");
            System.out.println("receive message:"+message);
        }
    };
    
    //监听队列
    //参数：String queue, boolean autoAck, Consumer callback
    /**
     * 参数明细：
     * 1、queue 队列名称
     * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
     * 3、callback，消费方法，当消费者接收到消息要执行的方法
     */
    channel.basicConsume(QUEUE,true,defaultConsumer);

## 一对多 Publish/Subscribe ##

配置单个交换机多个队列，将消息发给多个队列的消费方

（1）生产方：  
EXCHANGE\_FANOUT\_INFORM交换—（广播消息）–>QUEUE\_INFORM\_EMAIL和QUEUE\_INFORM\_SMS队列

try {
        //声明队列，如果队列在mq 中没有则要创建
        //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
        /**
         * 参数明细
         * 1、queue 队列名称
         * 2、durable 是否持久化，如果持久化，mq重启后队列还在
         * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
         * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
         * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
         */
        channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
        channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
        //声明一个交换机
        //参数：String exchange, String type
        /**
         * 参数明细：
         * 1、交换机的名称
         * 2、交换机的类型
         * fanout：对应的rabbitmq的工作模式是 publish/subscribe
         * direct：对应的Routing	工作模式
         * topic：对应的Topics工作模式
         * headers： 对应的headers工作模式
         */
        channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
        //进行交换机和队列绑定
        //参数：String queue, String exchange, String routingKey
        /**
         * 参数明细：
         * 1、queue 队列名称
         * 2、exchange 交换机名称
         * 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串
         */
        channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_FANOUT_INFORM,"");
        channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_FANOUT_INFORM,"");
        //发送消息
        //参数：String exchange, String routingKey, BasicProperties props, byte[] body
        /**
         * 参数明细：
         * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
         * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
         * 3、props，消息的属性
         * 4、body，消息内容
         */
        for(int i=0;i<5;i++){
            //消息内容
            String message = "send inform message to user";
            channel.basicPublish(EXCHANGE_FANOUT_INFORM,"",null,message.getBytes());
            System.out.println("send to mq "+message);
        }
    
    } catch (Exception e) {
        e.printStackTrace();
    } finally {
        //关闭连接
        //先关闭通道
        try {
            channel.close();
        } catch (IOException e) {
            e.printStackTrace();
        } catch (TimeoutException e) {
            e.printStackTrace();
        }
        try {
            connection.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

（2）消费方（QUEUE\_INFORM\_EMAIL）：

/**
     * 参数明细
     * 1、queue 队列名称
     * 2、durable 是否持久化，如果持久化，mq重启后队列还在
     * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
     * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
     * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
     */
    channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
    //声明一个交换机
    //参数：String exchange, String type
    /**
     * 参数明细：
     * 1、交换机的名称
     * 2、交换机的类型
     * fanout：对应的rabbitmq的工作模式是 publish/subscribe
     * direct：对应的Routing	工作模式
     * topic：对应的Topics工作模式
     * headers： 对应的headers工作模式
     */
    channel.exchangeDeclare(EXCHANGE_FANOUT_INFORM, BuiltinExchangeType.FANOUT);
    //进行交换机和队列绑定
    //参数：String queue, String exchange, String routingKey
    /**
     * 参数明细：
     * 1、queue 队列名称
     * 2、exchange 交换机名称
     * 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串
     */
    channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_FANOUT_INFORM, "");
    
    //实现消费方法
    DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
    
        /**
         * 当接收到消息后此方法将被调用
         * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
         * @param envelope 信封，通过envelope
         * @param properties 消息属性
         * @param body 消息内容
         * @throws IOException
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            //交换机
            String exchange = envelope.getExchange();
            //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
            long deliveryTag = envelope.getDeliveryTag();
            //消息内容
            String message= new String(body,"utf-8");
            System.out.println("receive message:"+message);
        }
    };
    
    //监听队列
    //参数：String queue, boolean autoAck, Consumer callback
    /**
     * 参数明细：
     * 1、queue 队列名称
     * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
     * 3、callback，消费方法，当消费者接收到消息要执行的方法
     */
    channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);

## 指定发送路由 Routing ##

配置单个交换机多个队列，通过指定key来路由到指定队列的消费方

（1）生产方：

try {
           //声明队列，如果队列在mq 中没有则要创建
           //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
           /**
            * 参数明细
            * 1、queue 队列名称
            * 2、durable 是否持久化，如果持久化，mq重启后队列还在
            * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
            * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
            * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
            */
           channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
           channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
           //声明一个交换机
           //参数：String exchange, String type
           /**
            * 参数明细：
            * 1、交换机的名称
            * 2、交换机的类型
            * fanout：对应的rabbitmq的工作模式是 publish/subscribe
            * direct：对应的Routing	工作模式
            * topic：对应的Topics工作模式
            * headers： 对应的headers工作模式
            */
           channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
           //进行交换机和队列绑定
           //参数：String queue, String exchange, String routingKey
           /**
            * 参数明细：
            * 1、queue 队列名称
            * 2、exchange 交换机名称
            * 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串
            */
           channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);
           channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_ROUTING_INFORM,"inform");
           channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS);
           channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_ROUTING_INFORM,"inform");
           //发送消息
           //参数：String exchange, String routingKey, BasicProperties props, byte[] body
           /**
            * 参数明细：
            * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
            * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
            * 3、props，消息的属性
            * 4、body，消息内容
            */
          /* for(int i=0;i<5;i++){
               //发送消息的时候指定routingKey
               String message = "send email inform message to user";
               channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL,null,message.getBytes());
               System.out.println("send to mq "+message);
           }
           for(int i=0;i<5;i++){
               //发送消息的时候指定routingKey
               String message = "send sms inform message to user";
               channel.basicPublish(EXCHANGE_ROUTING_INFORM,ROUTINGKEY_SMS,null,message.getBytes());
               System.out.println("send to mq "+message);
           }*/
           for(int i=0;i<5;i++){
               //发送消息的时候指定routingKey
               String message = "send inform message to user";
               channel.basicPublish(EXCHANGE_ROUTING_INFORM,"inform",null,message.getBytes());
               System.out.println("send to mq "+message);
           }
    
       } catch (Exception e) {
           e.printStackTrace();
       } finally {
           //关闭连接
           //先关闭通道
           try {
               channel.close();
           } catch (IOException e) {
               e.printStackTrace();
           } catch (TimeoutException e) {
               e.printStackTrace();
           }
           try {
               connection.close();
           } catch (IOException e) {
               e.printStackTrace();
           }
       }

（2）消费方：

/**
        * 参数明细
        * 1、queue 队列名称
        * 2、durable 是否持久化，如果持久化，mq重启后队列还在
        * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
        * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
        * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
        */
       channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
       //声明一个交换机
       //参数：String exchange, String type
       /**
        * 参数明细：
        * 1、交换机的名称
        * 2、交换机的类型
        * fanout：对应的rabbitmq的工作模式是 publish/subscribe
        * direct：对应的Routing	工作模式
        * topic：对应的Topics工作模式
        * headers： 对应的headers工作模式
        */
       channel.exchangeDeclare(EXCHANGE_ROUTING_INFORM, BuiltinExchangeType.DIRECT);
       //进行交换机和队列绑定
       //参数：String queue, String exchange, String routingKey
       /**
        * 参数明细：
        * 1、queue 队列名称
        * 2、exchange 交换机名称
        * 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串
        */
       channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_ROUTING_INFORM,ROUTINGKEY_EMAIL);
    
       //实现消费方法
       DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
    
           /**
            * 当接收到消息后此方法将被调用
            * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
            * @param envelope 信封，通过envelope
            * @param properties 消息属性
            * @param body 消息内容
            * @throws IOException
            */
           @Override
           public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
               //交换机
               String exchange = envelope.getExchange();
               //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
               long deliveryTag = envelope.getDeliveryTag();
               //消息内容
               String message= new String(body,"utf-8");
               System.out.println("receive message:"+message);
           }
       };
    
       //监听队列
       //参数：String queue, boolean autoAck, Consumer callback
       /**
        * 参数明细：
        * 1、queue 队列名称
        * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
        * 3、callback，消费方法，当消费者接收到消息要执行的方法
        */
       channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);

## 通配符匹配 Topic ##

声明一个交换机多个队列，通过配置带通配符的key来向不同的队列消费方发消息，规则如下：

*（星号）：可以（只能）匹配一个单词
    #（井号）：可以匹配多个单词（或者零个）

（1）生成方：

try {
           //建立新连接
           connection = connectionFactory.newConnection();
           //创建会话通道,生产者和mq服务所有通信都在channel通道中完成
           channel = connection.createChannel();
           //声明队列，如果队列在mq 中没有则要创建
           //参数：String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments
           /**
            * 参数明细
            * 1、queue 队列名称
            * 2、durable 是否持久化，如果持久化，mq重启后队列还在
            * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
            * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
            * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
            */
           channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
           channel.queueDeclare(QUEUE_INFORM_SMS,true,false,false,null);
           //声明一个交换机
           //参数：String exchange, String type
           /**
            * 参数明细：
            * 1、交换机的名称
            * 2、交换机的类型
            * fanout：对应的rabbitmq的工作模式是 publish/subscribe
            * direct：对应的Routing	工作模式
            * topic：对应的Topics工作模式
            * headers： 对应的headers工作模式
            */
           channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);
           //进行交换机和队列绑定
           //参数：String queue, String exchange, String routingKey
           /**
            * 参数明细：
            * 1、queue 队列名称
            * 2、exchange 交换机名称
            * 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串 ROUTINGKEY_EMAIL="inform.#.email.#"    ROUTINGKEY_SMS="inform.#.sms.#";
            */
           channel.queueBind(QUEUE_INFORM_EMAIL,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);
           channel.queueBind(QUEUE_INFORM_SMS,EXCHANGE_TOPICS_INFORM,ROUTINGKEY_SMS);
           //发送消息
           //参数：String exchange, String routingKey, BasicProperties props, byte[] body
           /**
            * 参数明细：
            * 1、exchange，交换机，如果不指定将使用mq的默认交换机（设置为""）
            * 2、routingKey，路由key，交换机根据路由key来将消息转发到指定的队列，如果使用默认交换机，routingKey设置为队列的名称
            * 3、props，消息的属性
            * 4、body，消息内容
            */
           for(int i=0;i<5;i++){
               //发送消息的时候指定routingKey
               String message = "send email inform message to user";
               channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.email",null,message.getBytes());
               System.out.println("send to mq "+message);
           }
           for(int i=0;i<5;i++){
               //发送消息的时候指定routingKey
               String message = "send sms inform message to user";
               channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms",null,message.getBytes());
               System.out.println("send to mq "+message);
           }
           for(int i=0;i<5;i++){
               //发送消息的时候指定routingKey
               String message = "send sms and email inform message to user";
               channel.basicPublish(EXCHANGE_TOPICS_INFORM,"inform.sms.email",null,message.getBytes());
               System.out.println("send to mq "+message);
           }
    
    
       } catch (Exception e) {
           e.printStackTrace();
       } finally {
           //关闭连接
           //先关闭通道
           try {
               channel.close();
           } catch (IOException e) {
               e.printStackTrace();
           } catch (TimeoutException e) {
               e.printStackTrace();
           }
           try {
               connection.close();
           } catch (IOException e) {
               e.printStackTrace();
           }
       }

（2）消费方：

/**
        * 参数明细
        * 1、queue 队列名称
        * 2、durable 是否持久化，如果持久化，mq重启后队列还在
        * 3、exclusive 是否独占连接，队列只允许在该连接中访问，如果connection连接关闭队列则自动删除,如果将此参数设置true可用于临时队列的创建
        * 4、autoDelete 自动删除，队列不再使用时是否自动删除此队列，如果将此参数和exclusive参数设置为true就可以实现临时队列（队列不用了就自动删除）
        * 5、arguments 参数，可以设置一个队列的扩展参数，比如：可设置存活时间
        */
       channel.queueDeclare(QUEUE_INFORM_EMAIL,true,false,false,null);
       //声明一个交换机
       //参数：String exchange, String type
       /**
        * 参数明细：
        * 1、交换机的名称
        * 2、交换机的类型
        * fanout：对应的rabbitmq的工作模式是 publish/subscribe
        * direct：对应的Routing	工作模式
        * topic：对应的Topics工作模式
        * headers： 对应的headers工作模式
        */
       channel.exchangeDeclare(EXCHANGE_TOPICS_INFORM, BuiltinExchangeType.TOPIC);
       //进行交换机和队列绑定
       //参数：String queue, String exchange, String routingKey
       /**
        * 参数明细：
        * 1、queue 队列名称
        * 2、exchange 交换机名称
        * 3、routingKey 路由key，作用是交换机根据路由key的值将消息转发到指定的队列中，在发布订阅模式中调协为空字符串   ROUTINGKEY_EMAIL="inform.#.email.#"
        */
       channel.queueBind(QUEUE_INFORM_EMAIL, EXCHANGE_TOPICS_INFORM,ROUTINGKEY_EMAIL);
    
       //实现消费方法
       DefaultConsumer defaultConsumer = new DefaultConsumer(channel){
    
           /**
            * 当接收到消息后此方法将被调用
            * @param consumerTag  消费者标签，用来标识消费者的，在监听队列时设置channel.basicConsume
            * @param envelope 信封，通过envelope
            * @param properties 消息属性
            * @param body 消息内容
            * @throws IOException
            */
           @Override
           public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
               //交换机
               String exchange = envelope.getExchange();
               //消息id，mq在channel中用来标识消息的id，可用于确认消息已接收
               long deliveryTag = envelope.getDeliveryTag();
               //消息内容
               String message= new String(body,"utf-8");
               System.out.println("receive message:"+message);
           }
       };
    
       //监听队列
       //参数：String queue, boolean autoAck, Consumer callback
       /**
        * 参数明细：
        * 1、queue 队列名称
        * 2、autoAck 自动回复，当消费者接收到消息后要告诉mq消息已接收，如果将此参数设置为tru表示会自动回复mq，如果设置为false要通过编程实现回复
        * 3、callback，消费方法，当消费者接收到消息要执行的方法
        */
       channel.basicConsume(QUEUE_INFORM_EMAIL,true,defaultConsumer);