公司网站建设外包流程图,答辩学网站开发知识能力要求,网站内容的创新,电商网站建设常见问题消费端ACK机制
在这之前已经完成了发送端的确认机制。可以保证数据成功的发送到RabbitMQ#xff0c;以及持久化机制#xff0c;然尔这依然无法完全保证整个过程的可靠性#xff0c;因为如果消息被消费过程中业务处理失败了#xff0c;但是消息却已经被标记为消费了以及持久化机制然尔这依然无法完全保证整个过程的可靠性因为如果消息被消费过程中业务处理失败了但是消息却已经被标记为消费了如果又没有任何重度机制那结果基本等于丢消息。在消费端如何保证消息不丢呢
在rabbitMQ的消费端会有ACK机制。即消费端消费消息后需要发送ACK确认报文给Broker端告知自己是否已经消费完成否则可能会一直重发消息直到消息过期AUTO模式。同时这个也是最终一致性、可恢复性的基础。一般有如下手段采用NONE模式消费的过程中自行捕捉异常引发异常后直接记录日志并落到异常处理表再通过后台定时任务扫描异常恢复表做重度动作。如果业务不自行处理则有丢失数据的风险。采用AUTO自动ACK模式不主动捕获异常当消费过程中出现异常时会将消息放回Queue中然后消息会被重新分配到其他消费节点如果没有则还是选择当前节点重新被消费默认会一直重发消息并直到消费完成返回ACK或者一直到过期。采用MANUAL手动ACK模式消费者自行控制流程并手动调用channel相关的方法返回ACK。 7.6.1 手动ACK机制-Reject
maven导入 dependencygroupIdcom.rabbitmq/groupIdartifactIdamqp-client/artifactIdversion5.9.0/version/dependency生产者
import com.rabbitmq.client.BuiltinExchangeType;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;import java.nio.charset.StandardCharsets;public class Product {public static void main(String[] args) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setUri(amqp://root:123456node1:5672/%2f);try (Connection connection factory.newConnection();Channel channel connection.createChannel(); ) {// 定义交换器队列channel.exchangeDeclare(ack.ex, BuiltinExchangeType.DIRECT, false, false, false, null);// 定义队列channel.queueDeclare(ack.qu, false, false, false, null);// 队列绑定channel.queueBind(ack.qu, ack.ex, ack.rk);for (int i 0; i 5; i) {byte[] sendBytes (hello- i).getBytes(StandardCharsets.UTF_8);channel.basicPublish(ack.ex, ack.rk, null, sendBytes);}} catch (Exception e) {e.printStackTrace();}}
}
消费者
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setUri(amqp://root:123456node1:5672/%2f);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(ack.qu, false, false, false, null);DefaultConsumer consumer new DefaultConsumer(channel) {Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println(确认的消息内容: new String(body));// 找收消息// Nack与Reject的区别在于nack可以对多条消息进行拒收而reject只能拒收一条。// requeue为true表示不确认的消息会重新放回队列。channel.basicReject(envelope.getDeliveryTag(), true);}};channel.basicConsume(ack.qu,// 非自动确认false,// 消费者的标签ack.consumer,// 回调函数consumer);}
}
发送测试
首先执行生产者向队列中发送数据。然后执行消费者检查拒收的处理。
在消费者的控制台将持续不断的输出消息信息
确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4
确认的消息内容:hello-0
确认的消息内容:hello-1
......
确认的消息内容:hello-0按照发送的顺序将不断的被打印。
那此时消息是什么状态呢查看下消息队列中的信息
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 5 │ 5 │ 1 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]# 可以看到当前的消息处于unack的状态。由于消息被不断的重新放回队列而消费者又只有当前这一个所以在不断拒收中被放回。
那如果将消息拒绝改为不重新放回队列会如何呢来验证下。
首先修改消费者的代码:
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setUri(amqp://root:123456node1:5672/%2f);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(ack.qu, false, false, false, null);DefaultConsumer consumer new DefaultConsumer(channel) {Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println(确认的消息内容: new String(body));// 找收消息// Nack与Reject的区别在于nack可以对多条消息进行拒收而reject只能拒收一条。// requeue为false表示不确认的消息不会重新放回队列。//channel.basicReject(envelope.getDeliveryTag(), true);channel.basicReject(envelope.getDeliveryTag(), false);}};channel.basicConsume(ack.qu,// 非自动确认false,// 消费者的标签ack.consumer,// 回调函数consumer);}
}
再次执行消费者。
确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4而这一次消息没有再循环打印。只输出一遍再检查下消息在队列中的状态
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 0 │ 0 │ 1 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]# 通过观察发现消息已经没有在队列中了那就是消息已经被丢弃了。
7.6.2 手动ACK机制-ack
消费者修改为ACK确认处理
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setUri(amqp://root:123456node1:5672/%2f);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(ack.qu, false, false, false, null);DefaultConsumer consumer new DefaultConsumer(channel) {Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println(确认的消息内容: new String(body));// 消息确认并且非批量确认multiple为false表示只确认了单条channel.basicAck(envelope.getDeliveryTag(), false);}};channel.basicConsume(ack.qu,// 非自动确认false,// 消费者的标签ack.consumer,// 回调函数consumer);}
}
此时可以先运行消息者。等待消息推送。然后运行生产者将消息推送此时便可以看到消费者的控制台输出
确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4观察队列中的信息
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 0 │ 0 │ 1 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]# 在队列中消息已经被成功的消费了。
7.6.3 手动ACK机制-nack
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import java.io.IOException;public class Consumer {public static void main(String[] args) throws Exception {ConnectionFactory factory new ConnectionFactory();factory.setUri(amqp://root:123456node1:5672/%2f);Connection connection factory.newConnection();Channel channel connection.createChannel();channel.queueDeclare(ack.qu, false, false, false, null);DefaultConsumer consumer new DefaultConsumer(channel) {Overridepublic void handleDelivery(// 消费者标签String consumerTag,// 消费者的封装Envelope envelope,// 消息属性AMQP.BasicProperties properties,// 消息体byte[] body)throws IOException {System.out.println(确认的消息内容: new String(body));// 消息批量不确认即批量丢弃每5个做一次批量消费// 参数1消息的标签// multiple为false 表示不确认当前是一个消息。true就是多个消息。// requeue为true表示不确认的消息会重新放回队列。// 每5条做一次批量确认,_deliveryTag从1开始if (envelope.getDeliveryTag() % 5 0) {System.out.println(批量确认执行);channel.basicNack(envelope.getDeliveryTag(), true, false);}}};channel.basicConsume(ack.qu,// 非自动确认false,// 消费者的标签ack.consumer,// 回调函数consumer);}
}
执行消费者程序,然后再执行生产者。查看消费端的控制台
确认的消息内容:hello-0
确认的消息内容:hello-1
确认的消息内容:hello-2
确认的消息内容:hello-3
确认的消息内容:hello-4
批量确认执行由于此处采用的是不重新放回队列所以数据接收到之后被丢弃了。
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 0 │ 0 │ 0 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘队列中的数据也已经被处理掉了。
7.6.4 手动ACK机制-SpringBoot
首先是Maven导入 dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactIdversion2.2.8.RELEASE/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-web/artifactIdversion2.2.8.RELEASE/version/dependency配制文件application.yml
spring:application:name: consumer-ackrabbitmq:host: node1port: 5672virtual-host: /username: rootpassword: 123456# 配制消费端ack信息。listener:simple:acknowledge-mode: manual# 重试超过最大次数后是否拒绝default-requeue-rejected: falseretry:# 开启消费者重度(false时关闭消费者重试false不是不重试而是一直收到消息直到ack确认或者一直到超时enable: true# 最大重度次数max-attempts: 5# 重试间隔时间(单位毫秒)initial-interval: 1000
主启动类
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;import java.nio.charset.StandardCharsets;SpringBootApplication
public class Main {Autowired private RabbitTemplate rabbitTemplate;public static void main(String[] args) {SpringApplication.run(Main.class, args);}/*** 在启动后就开始向MQ中发送消息** return*/Beanpublic ApplicationRunner runner() {return args - {Thread.sleep(5000);for (int i 0; i 10; i) {MessageProperties props new MessageProperties();props.setDeliveryTag(i);Message message new Message((消息: i).getBytes(StandardCharsets.UTF_8), props);rabbitTemplate.convertAndSend(ack.ex, ack.rk, message);}};}
}当主类启动后会延迟5秒向MQ中发送10条记录。
队列配制
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Exchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;Configuration
public class RabbitConfig {Beanpublic Queue queue() {return new Queue(ack.qu, false, false, false, null);}Beanpublic Exchange exchange(){return new DirectExchange(ack.ex,false,false,null);}Beanpublic Binding binding(){return BindingBuilder.bind(queue()).to(exchange()).with(ack.rk).noargs();}
}
使用推送模式来查确认消息
监听器MQ队列推送消息至listener
import com.rabbitmq.client.Channel;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;import java.io.IOException;
import java.util.concurrent.ThreadLocalRandom;Component
public class MessageListener {/*** NONE模式则只要收到消息后就立即确认消息出列标识已消费有丢数据风险** pAUTO模式看情况确认如果此时消费者抛出异常则消息会返回队列中** pWANUAL模式需要显示的调用当前channel的basicAck方法** param channel* param deliveryTag* param msg*/// RabbitListener(queues ack.qu, ackMode AUTO)// RabbitListener(queues ack.qu, ackMode NONE)RabbitListener(queues ack.qu, ackMode MANUAL)public void handMessageTopic(Channel channel, Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag, Payload String msg) {System.out.println(消息内容 msg);ThreadLocalRandom current ThreadLocalRandom.current();try {if (current.nextInt(10) % 3 ! 0) {// 手动nack告诉broker消费者处理失败最后一个参数表示是否需要将消息重新入列// channel.basicNack(deliveryTag, false, true);// 手动拒绝消息第二个参数表示是否重新入列channel.basicReject(deliveryTag, true);} else {// 手动ACKdeliveryTag表示消息的唯一标志multiple表示是否批量确认channel.basicAck(deliveryTag, false);System.out.println(已经确认的消息 msg);}Thread.sleep(ThreadLocalRandom.current().nextInt(500, 3000));} catch (IOException e) {e.printStackTrace();} catch (InterruptedException e) {}}
}消息有33%的概率被拒绝这样又会被重新放回队列等待下次推送。
启动测试
运行main方法
【确认】消息内容消息:0
【拒绝】消息内容消息:1
【拒绝】消息内容消息:2
【拒绝】消息内容消息:3
【确认】消息内容消息:4
【确认】消息内容消息:5
【拒绝】消息内容消息:6
【拒绝】消息内容消息:7
【拒绝】消息内容消息:8
【拒绝】消息内容消息:9
【确认】消息内容消息:1
【拒绝】消息内容消息:2
【拒绝】消息内容消息:3
【拒绝】消息内容消息:6
【确认】消息内容消息:7
【确认】消息内容消息:8
【拒绝】消息内容消息:9
【拒绝】消息内容消息:2
【拒绝】消息内容消息:3
【拒绝】消息内容消息:6
【确认】消息内容消息:9
【确认】消息内容消息:2
【拒绝】消息内容消息:3
【拒绝】消息内容消息:6
【确认】消息内容消息:3
【拒绝】消息内容消息:6
【确认】消息内容消息:6从观察到的结果也印证了反复的被推送接收的一个过程中使用命令查看队列的一个消费的情况
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 6 │ 6 │ 1 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 1 │ 1 │ 1 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 0 │ 0 │ 1 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘使用拉确认消息
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.GetResponse;
import org.springframework.amqp.rabbit.core.ChannelCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;import java.nio.charset.StandardCharsets;
import java.util.concurrent.ThreadLocalRandom;RestController
public class MsgController {Autowired private RabbitTemplate rabbitTemplate;RequestMapping(/msg)public String getMessage() {String message rabbitTemplate.execute(new ChannelCallbackString() {Overridepublic String doInRabbit(Channel channel) throws Exception {GetResponse getResponse channel.basicGet(ack.qu, false);if (null getResponse) {return 你已经消费完所有的消息;}String message new String(getResponse.getBody(), StandardCharsets.UTF_8);if (ThreadLocalRandom.current().nextInt(10) % 3 0) {// 执行消息确认操作channel.basicAck(getResponse.getEnvelope().getDeliveryTag(), false);return 已确认的消息: message;} else {// 拒收一条消息并重新放回队列channel.basicReject(getResponse.getEnvelope().getDeliveryTag(), true);return 拒绝的消息: message;}}});return message;}
}
在浏览器中访问同样有66%的概率会被拒绝仅33%会被确认。
注如果与监听在同一个工程需将监听器给注释。
启动main函数在浏览器中访问。http://127.0.0.1:8080/msg
可以看到返回:
拒绝的消息:消息:0
已确认的消息:消息:1
拒绝的消息:消息:2
......
已确认的消息:消息:9
你已经消费完所有的消息同样的观察队列的一个消费情况
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 8 │ 0 │ 8 │ 0 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 3 │ 0 │ 3 │ 0 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]# rabbitmqctl list_queues name,messages_ready,messages_unacknowledged,messages,consumers --formatter pretty_table
Timeout: 60.0 seconds ...
Listing queues for vhost / ...
┌───────────────┬────────────────┬─────────────────────────┬──────────┬───────────┐
│ name │ messages_ready │ messages_unacknowledged │ messages │ consumers │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ ack.qu │ 0 │ 0 │ 0 │ 0 │
├───────────────┼────────────────┼─────────────────────────┼──────────┼───────────┤
│ persistent.qu │ 1 │ 0 │ 1 │ 0 │
└───────────────┴────────────────┴─────────────────────────┴──────────┴───────────┘
[rootnullnull-os rabbitmq]#
使用拉模式进行消息ACK确认也已经完成。
