RabbitMQ不讲武德，发个消息也这么多花招(1)

*/

@rabbithandler

@rabbitlistener(queues = “longly_wolf_product_queue”)

public void fanoutconsumer(string msg){

system.out.println(“fanout交换机收到消息：” + msg);

}

}

• 4、新增一个 rabbitexchangecontroller 类来作为生产者，进行消息发送：

@restcontroller

@requestmapping(“/exchange”)

public class rabbitexchangecontroller {

@autowired

private rabbittemplate rabbittemplate;

@getmapping(value=“/send/direct”)

public string senddirect(string routingkey,@requestparam(value = “msg”,defaultvalue = “no direct message”) string msg){

rabbittemplate.convertandsend(“longly_wolf_direct_exchange”,routingkey,msg);

return “succ”;

}

@getmapping(value=“/send/topic”)

public string sendtopic(string routingkey,@requestparam(value = “msg”,defaultvalue = “no topic message”) string msg){

rabbittemplate.convertandsend(“longly_wolf_topic_exchange”,routingkey,msg);

return “succ”;

}

@getmapping(value=“/send/fanout”)

public string sendfaout(string routingkey,@requestparam(value = “msg”,defaultvalue = “no faout message”) string msg){

rabbittemplate.convertandsend(“longly_wolf_fanout_exchange”,routingkey,msg);

return “succ”;

}

}

• 5、启动服务，当我们调用第一个接口时候，路由键和绑定键 order.detail 精确匹配时，directconsumer 就会收到消息，同样的，调用第二接口时，路由键满足 user.# 时，topicconsumer 就会收到消息，而只要调用第三个接口，不论是否指定路由键，fanoutconsumer 都会收到消息。

消息过期了怎么办

=====================================================================

简单的发送消息我们学会了，难道这就能让我们就此止步了吗？显然是不能的，要玩就要玩高级点，所以接下来让我们给消息加点佐料。

ttl（time-to-live）

---

ttl 即 一条消息在队列中的最大存活时间。在一条在队列中超过配置的 ttl 的消息称为已死消息。但是需要注意的是，已死消息并不能保证会立即从队列中删除，但是能保证已死的消息不会被投递出去。

设置 ttl 的方式有两种：

• 1、给队列设置 x-message-ttl，此时所有被投递到队列中的消息，都会在到达 ttl 时成为已死消息。

这种情况就会出现当一条消息同时路由到 n 个带有 ttl 时间的队列，而由于每个队列的 ttl 不一定相同，所以同一条消息在不同的队列中可能会在不同时间死亡或者不会死亡（未设置 ttl ），所以一个队列中的消息死亡不会影响到其他队列中的消息。

• 2、单独给某一条消息设置过期时间。

此时需要注意的时，当消息达到 ttl 时，可能不会马上被丢弃，因为只有处于队列头部消息过期后才会被丢弃，假如队列头部的消息没有设置 ttl，而第 2 条消息设置了 ttl，那么即使第 2 条消息成为了已死消息，也必须要等到队列头部的消息被消费之后才会被丢弃，而已死消息在被丢弃之前也会被计入统计数据（比如队列中的消息总数）。所以为了更好的利用 ttl 特性，建议让消费者在线消费消息，这样才能确保消息更快的被丢弃，防止消息堆积。

ps：消息过期和消费者传递之间可能存在自然的竞争条件。例如，消息可能在发送途中（未到达消费者）过期。

队列的生存

---

和 ttl 针对消息不同的是，我们可以通过设置过期时间属性 `x-expires`` 来处理队列，当在指定过期时间内内未使用队列时，服务器保证将删除队列（但是无法保证在过期时间过后队列将以多快的速度被删除）。

ttl 和过期时间实战

---

• 1、在上面定义的 rabbitconfig 类中，再新增一个 ttl 队列并将其绑定到 direct 交换机上：

@bean(“ttlqueue”)

public queue ttlqueue(){

map<string, object> map = new hashmap<string, object>();

map.put(“x-message-ttl”, 5000);//队列中所有消息5秒后过期

map.put(“x-expires”, 100000);//队列闲置10秒后被删除

//参数1-name：队列名称

//参数2-durable：是否持久化

//参数3-exclusive:是否排他。设置为true时，则该队列只对声明当前队列的连接(connection)可用,一旦连接断开，队列自动被删除

//参数4-autodelete：是否自动删除。前提是必须要至少有一个消费者先连上当前队列，然后当所有消费者都断开连接之后，队列自动被删除

return new queue(“longly_wolf_ttl_queue”,false,false,false,map);

}

//ttl队列绑定到direct交换机（交换机和队列可以多对多）

@bean

public binding ttlbindfanoutexchange(@qualifier(“ttlqueue”) queue queue, @qualifier(“directexchange”) directexchange directexchange){

return bindingbuilder.bind(queue).to(directexchange).with(“test.ttl”);

}

• 2、在 exchangeconsumer 消费者类上监听 ttl 队列（和其他消费者不同的时候，这里为了打印出队列属性，改成了通过 message 对象来接收消息 ）：

/**

• 监听ttl消息队列

*/

@rabbithandler

@rabbitlistener(queues = “longly_wolf_ttl_queue”)

public void ttlconsumer(message message){

system.out.println(“ttl队列收到消息：” + new string(message.getbody()));

system.out.println(“ttl队列收到消息：” + jsonobject.tojsonstring(message.getmessageproperties()));

}

• 3、在生产者类 rabbitexchangecontroller 上新增一个接口用来测试发送过期消息，这里通过 messageproperties 设置的 expiration 属性就相当于是给单条消息设置了一个 ttl ：

@getmapping(value=“/send/ttl”)

public string sendttl(string routingkey,@requestparam(value = “msg”,defaultvalue = “no ttl message”) string msg){

messageproperties messageproperties = new messageproperties();

messageproperties.setexpiration(“5000”);//5秒后被删除，即ttl属性（针对单条消息）

message message = new message(msg.getbytes(), messageproperties);

rabbittemplate.convertandsend(“longly_wolf_direct_exchange”,routingkey,message);

return “succ”;

}

• 4、此时如果我们把消费者的监听去掉之后再发送消息，在管理后台就可以看到 5 秒之后消息会被删除，10 秒之后队列会被删除。

ps：如果同时给队列和单条消息都设置了 ttl，则会以时间短的为主。

其他属性

---

队列中还有其他一些属性可以设置，在这里我们就不一一举例了：

• x-message-ttl：队列中消息的存活时间（毫秒），达到ttl的消息可能会被删除。

• x-expires：队列在多长时间（毫秒）没有被访问以后会被删除。

• x-max-length：队列中的最大消息数。

• x-max-length-bytes：队列的最大容量（bytes）。

• overflow：队列溢出之后的策略。主要可以配置如下参数：reject-publish - 直接丢弃最近发布的消息，如若启用了publisher confirm（发布者确认），发布者将通过发送 basic.nack 消息通知拒绝，如果当前队列绑定有多个消费者，则消息在收到 basic.nack 拒绝通知后，仍然会被发布到其他队列；drop-head - 丢弃队列头部消息（集群模式下只支持这种策略） reject-publish-dlx - 最近发布的消息会进入死信队列。

• x-dead-letter-exchange：队列的死信交换机。

• x-dead-letter-routing-key：死信交换机的路由键。

• x-single-active-consumer：true/false。表示是否最多只允许一个消费者消费，如果有多个消费者同时绑定，则只会激活第一个，除非第一个消费者被取消或者死亡，才会自动转到下一个消费者。

• x-max-priority：队列中消息的最大优先级， 消息的优先级不能超过它。

• x-queue-mode：3.6.0 版本引入的，主要是为了实现惰性加载。队列将收到的消息尽可能快的进行持久化操作到磁盘上，然后只有在用户请求的时候才会加载到 ram 内存。这个参数支持两个值：default 和 lazy。当不进行设置的时候，就是默认为 default，不做任何改变；当设置为 lazy 就会进行懒加载。

• x-queue-master-locator：为了保证消息的 fifo，所以在高可用集群模式下需要选择一个节点作为主节点。这个参数主要有三种模式：min-masters- 托管最小数量的绑定主机的节点；client-local- 选择声明的队列已经连接到客户端的节点；random- 随机选择一个节点。

神奇的死信队列（dead letter）

=================================================================================

上面的参数介绍中，提到了死信队列，这又是什么新鲜的东西呢？其实从名字上来看很好理解，就是指的已死的消息，或者说无家可归的消息。一个消息进入死信队列，主要有以下三种条件：

• 1、消息被消费者拒绝并且未设置重回队列。

• 2、消息过期（即设置了 ttl）。

• 3、队列达到最大长度，超过了 max length 或 max length bytes，则队列头部的消息会被发送到死信队列。

死信队列实战

---

• 1、在上面定义的 rabbitconfig 类中，定义一个死信交换机，并将之前的 ttl 队列新增一个属性 x-dead-letter-exchange，最后再将死信队列和死信交换机进行绑定：

//直连死信交换机(也可以用topic或者fanout类型交换机)

@bean(“deatletterexchange”)

public directexchange deatletterexchange(){

return new directexchange(“longly_wolf_dead_letter_direct_exchange”);

}

@bean(“ttlqueue”)

public queue ttlqueue(){

map<string, object> map = new hashmap<string, object>();

map.put(“x-message-ttl”, 5000);//队列中所有消息5秒后过期

map.put(“x-dead-letter-exchange”, “longly_wolf_dead_letter_direct_exchange”);//已死消息会进入死信交换机

return new queue(“longly_wolf_ttl_queue”,false,false,false,map);

}

//死信队列

@bean(“deadletterqueue”)

public queue deadletterqueue(){

return new queue(“longly_wolf_dead_letter_queue”);

}

• 2、在 exchangeconsumer 消费者类上将监听 ttl 队列的监听取消，注释掉监听：

/**

• 监听ttl消息队列

*/

@rabbithandler

// @rabbitlistener(queues = “longly_wolf_ttl_queue”)

public void ttlconsumer(message message){

system.out.println(“ttl队列收到消息：” + new string(message.getbody()));

system.out.println(“ttl队列收到消息：” + jsonobject.tojsonstring(message.getmessageproperties()));

}

• 3、此时 ttl 队列无消费者，并且设置了消息的 ttl 为 5 秒，所以 5 秒之后就会进入死信队列。

• 5、访问接口：http://localhost:8080/exchange/send/ttl?routingkey=test&msg=测试死信队列，发送消息之后，等待 5 秒就查看消息，进入死信队列：

消息真的发送成功了吗

=======================================================================

了解了消息的基本发送功能之后，就可以高枕无忧了吗？消息发出去之后，消费者真的收到消息了吗？消息发送之后如何知道消息发送成功了？假如发送消息路由错了导致无法路由到队列怎么办？大家是不是都有这些疑问呢？别着急，接下来就让我们来一一来分析一下。

一条消息从生产者开始发送消息到消费者消费完消息主要可以分为以下 4 个阶段：

• 1、生产者将消息发送到 broker （即：rabbitmq 的交换机）。

• 2、交换机将消息路由到队列。

• 3、队列收到消息后存储消息。

• 4、消费者从队列获取消息进行消费。

接下来我们就从这 4 个步骤上来逐步分析 rabbitmq 如何保证消息发送的可靠性。

消息真的到达交换机了吗

---

当我们发送一条消息之后，如何知道对方收到消息了？这就和我们写信一样，写一封信出去，如何知道对方收到我们寄出去的信？最简单的方式就是对方也给我们回一封信，我们收到对方的回信之后就可以知道自己的信已经成功寄达。

在 rabbitmq 中服务端也提供了 2 种方式来告诉客户端（生产者）是否收到消息：transaction（事务）模式和 confirm（确认）模式。

transaction（事务） 模式

在 java api 编程中开启事务只需要增加以下代码即可：

try {

channel.txselect();//开启事务

channel.basicpublish(“”, queue_name, null, msg.getbytes());

channel.txcommit();//提交事务

}catch (exception e){

channel.txrollback();//消息回滚

}

在 spring boot 中需要对 rabbittemplate 进行事务设置：

@bean

public rabbittemplate rabbittemplate(connectionfactory connectionfactory){

rabbittemplate rabbittemplate = new rabbittemplate();

rabbittemplate.setconnectionfactory(connectionfactory);

rabbittemplate.setchanneltransacted(true);//开启事务

return rabbittemplate;

}

为了了解 rabbitmq 当中事务机制的原理，我们在 wireshark 中输入 ip.addr==192.168.1.1 对本地 ip 进行抓包，发送一条消息之后，抓到如下数据包：

通过数据包，可以得出开启事务之后，除了原本的发送消息之外，多出了开启事务和事务提交的通信：

开启事务之后，有一个致命的缺点就是发送消息流程会被阻塞。也就是说必须一条消息发送成功之后，才会允许发送另一条消息。正因为事务模式有这个缺点，所以一般情况下并不建议在生产环境开启事务，那么有没有更好的方式来实现消息的送达确认呢？那么就让我们再看看confirm（确认）模式。

confirm（确认）模式

消息确认模式又可以分为三种（事务模式和确认模式无法同时开启）：

• 单条确认模式：发送一条消息，确认一条消息。此种确认模式的效率也不高。

• 批量确认模式：发送一批消息，然后同时确认。批量发送有一个缺点就是同一批消息一旦有一条消息发送失败，就会收到失败的通知，需要将这一批消息全部重发。

• 异步确认模式：一边发送一边确认，消息可能被单条确认也可能会被批量确认。

java api 实现确认模式

• 单条消息确认模式

channel.confirmselect();//开启确认模式

channel.basicpublish(“”,queue_name,null,msg.getbytes());

if (channel.waitforconfirms()){//wait.forconfirms(long time)方法可以指定等待时间

system.out.println(“消息确认发送成功”);

}

• 批量确认模式

channel.confirmselect();//开启确认模式

//批量发送

for (int i=0;i<10;i++){

channel.basicpublish(“”,queue_name,null,msg.getbytes());

}

try{

channel.waitforconfirmsordie();

}catch (ioexception e){//只要有1条消息未被确认，就会抛出异常

system.out.println(“有消息发送失败了”);

}

• 异步确认模式

channel.addconfirmlistener(new confirmlistener() {

/**

• 已确认消息，即发送成功后回调

• @param deliverytag -唯一标识id（即发送消息时获取到的nextpublishseqno）

• @param multiple - 是否批量确认，当multiple=true，表示<=deliverytag的消息被批量确认，multiple=false，表示只确认了单条

*/

@override

public void handleack(long deliverytag, boolean multiple) throws ioexception {//成功回调

system.out.println(“收到确认消息了”);

//todo 可以做一些想做的事

}

/**

• 发送失败消息后回调

• @param deliverytag -唯一标识id（即发送消息时获取到的nextpublishseqno）

• @param multiple - 是否批量确认，当multiple=true，表示<=deliverytag的消息被批量确认，multiple=false，表示只确认了单条

*/

@override

public void handlenack(long deliverytag, boolean multiple) throws ioexception {//失败回调

if (multiple) {//批量确认，<deliverytag的消息都发送失败

//todo 消息重发？

} else {//非批量，=deliverytag的消息发送失败

//todo 消息重发？

}

}

});

channel.confirmselect();//开启确认模式

for (int i=0;i<10;i++){//批量发送

long nextseqno = channel.getnextpublishseqno();//获取发送消息的唯一标识(从1开始递增)

//todo 可以考虑把消息id存起来

channel.basicpublish(“”,queue_name,null,msg.getbytes());

}

springboot 实现确认模式

通过配置文件 spring.rabbitmq.publisher-confirm-type 参数进行配置确认（旧版本是 spring.rabbitmq.publisher-confirms 参数）。

• 1、新增配置文件属性配置

spring:

rabbitmq:

publisher-confirm-type: correlated # none-表示禁用回调(默认) simple- 参考rabbitexchangecontroller#sendwithsimpleconfirm()方法

• 2、rabbitconfig 配置文件中修改如下：

@bean

public rabbittemplate rabbittemplate(connectionfactory connectionfactory){

rabbittemplate rabbittemplate = new rabbittemplate();

rabbittemplate.setconnectionfactory(connectionfactory);

// rabbittemplate.setchanneltransacted(true);//开启事务

//消息是否成功发送到exchange

rabbittemplate.setconfirmcallback(new rabbittemplate.confirmcallback() {

@override

public void confirm(correlationdata correlationdata, boolean ack, string cause) {

if (!ack){//消息发送失败

system.out.println(“消息发送失败，原因为：” + cause);

return;

}

//消息发送成功

system.out.println(“消息发送成功”);

}

});

return rabbittemplate;

}

这样当我们发送消息成功之后，就会收到回调。

• 3、当上面的参数配置修改为 simple，则需要在发送消息的时候使用 invoke 调用 waitforconfirms 或者 waitforconfirmsordie 方法来确认是否发送成功：

@getmapping(value=“/send/confirm”)

public string sendwithsimpleconfirm(string routingkey,@requestparam(value = “msg”,defaultvalue = “no direct message”) string msg){

//使用waitforconfirms方法确认

boolean sendflag = rabbittemplate.invoke(operations -> {

rabbittemplate.convertandsend(

“longly_wolf_direct_exchange”,

“routingkey”,

msg

);

return rabbittemplate.waitforconfirms(5000);

});

//也可以使用waitforconfirmsordie方法确认

boolean sendflag2 = rabbittemplate.invoke(operations -> {

rabbittemplate.convertandsend(

“longly_wolf_direct_exchange”,

“routingkey”,

msg

);

try {

rabbittemplate.waitforconfirmsordie(5000);

}catch (exception e){

return false;

}

return true;

});

system.out.println(sendflag);

system.out.println(sendflag2);

return “succ”;

}

消息无法从交换机路由到正确的队列怎么办

---

上面通过事务或者确认机制确保了消息成功发送到交换机，那么接下来交换机会负责将消息路由到队列，这时候假如队列不存在或者路由错误就会导致消息路由失败，这又该如何保证呢？

同样的，rabbitmq 中也提供了 2 种方式来确保消息可以正确路由到队列：开启监听模式或者通过新增备份交换机模式来备份数据。

监听回调

上面介绍的是消息是否发送到交换机的回调，而从交换机路由到队列，同样可以开启确认模式。

java api 方式开启监听模式

下面就是开启监听主要代码，为了节省篇幅，省略了其余不相干代码（完成代码已上传至 github）

channel.addreturnlistener(new returnlistener() {

@override

public void handlereturn(int replycode, string replytext, string exchange, string routingkey, amqp.basicproperties properties, byte[] body) throws ioexception {

system.out.println(“收到未路由到队列的回调消息：” + new string(body));

}

});

//注意这里的第三个参数，mandatory需要设置为true（发送一个错误的路由，即可收到回调）

channel.basicpublish(exchange_name,“error_routing_key”,true,null,msg.getbytes());

spring boot 开启监听模式

在 rabitconfig 类中添加如下配置：

@bean

public rabbittemplate rabbittemplate(connectionfactory connectionfactory){

rabbittemplate rabbittemplate = new rabbittemplate();

rabbittemplate.setconnectionfactory(connectionfactory);

rabbittemplate.setmandatory(true);//开启监听回调

//消息是否成功被路由到队列，没有路由到队列时会收到回调（原setreturncallback在2.0版本已过期）

rabbittemplate.setreturnscallback(new rabbittemplate.returnscallback() {

@override

public void returnedmessage(returnedmessage returnedmessage) {

system.out.println(“收到未路由到队列的回调消息：” + new string(returnedmessage.getmessage().getbody()));

}

});

return rabbittemplate;

}

备份交换机

除了开启监听的方式，还可以通过定义备份交换机的方式来实现，当原交换机无法正确路由到队列时，则会进入备份交换机，再由备份交换机路由到正确队列（要注意区分备份交换机和死信交换机的区别）。

java api 实现备份交换机

下面就是一个实现备份交换机的例子，因为这里备份交换机定义的是 topic 类型，所有路由必须满足定义好的路由，实际使用中一般会设置会 fanout，因为无法预测错误的路由到底是多少：

//声明交换机且指定备份交换机

map<string,object> argmap = new hashmap<string,object>();

argmap.put(“alternate-exchange”,“test_alternate_exchange”);

channel.exchangedeclare(exchange_name, builtinexchangetype.direct,false,false,argmap);

//队列和交换机进行绑定

channel.queuebind(queue_name,exchange_name,routeing_key);

//声明备份交换机和备份队列，并绑定（为了防止收不到消息，备份交换机一般建议设置为fanout类型）

channel.queuedeclare(“bak_queue”, false, false, false, null);

channel.exchangedeclare(“test_alternate_exchange”, builtinexchangetype.topic);

channel.queuebind(“bak_queue”,“test_alternate_exchange”,“error.#”);

string msg = “i’m a bak exchange msg”;

channel.basicpublish(exchange_name,“error.routing_key”,null,msg.getbytes());

spring boot 实现备份交换机

spring boot 实现备份交换机原理和 java api 实现相同：

• 1、首先在 rabbiconfig 中新增两个交换机，一个是原始交换机，一个是备份交换机，同时新增一个备份队列和备份交换机进行绑定，这里的备份交换机是一个 fanout 类型，注意因为这里主要是演示备份交换机，所以这里的原始交换机没有和任何队列绑定，也就无法路由到队列，从而使得消息进入备份交换机：

//用于测试备份交换机的原直连交换机

@bean(“bakdirectechange”)

public directexchange bakdirectechange(){

map argmap = new hashmap<>();

argmap.put(“alternate-exchange”, “longly_wolf_bak_fanout_exchange”);

return new directexchange(“longly_wolf_bak_origin_direct_exchange”,false,false,argmap);

}

//备份广播交换机

@bean(“bakfanoutexchange”)

public fanoutexchange bakfanoutexchange(){

return new fanoutexchange(“longly_wolf_bak_fanout_exchange”);

}

//备份队列

@bean(“bakqueue”)

public queue bakqueue(){

return new queue(“lonely_wolf_bak_queue”);

}

//备份交换机和备份队列进行绑定

@bean

public binding bindexchange(@qualifier(“bakqueue”) queue queue, @qualifier(“bakfanoutexchange”) fanoutexchange fanoutexchange){

return bindingbuilder.bind(queue).to(fanoutexchange);

}

2、在消费者类 exchangeconsumer 中监听备份队列：

/**

• 监听备份消息队列

*/

@rabbithandler

@rabbitlistener(queues = “lonely_wolf_bak_queue”)

public void bakqueueconsumer(message message){

system.out.println(“备份队列收到消息：” + new string(message.getbody()));

}

• 3、最后在生产者类 rabbitexchangecontroller 中新增一个消息发送的方法进行消息发送：

@getmapping(value=“/send/bak”)

public string sendbak(string routingkey,@requestparam(value = “msg”,defaultvalue = “no bak message”) string msg){

rabbittemplate.convertandsend(“longly_wolf_bak_origin_direct_exchange”,routingkey,msg);

return “succ”;

}

调用之后可以看到，备份队列会收到消息，从而说明了消息在无法路由到队列时会进入到备份队列。

队列存储消息后发生异常怎么办

---

在保证了前面两个阶段的可靠性之后，消息终于安全抵达了队列，那么这时候就绝对安全了吗？

当我们的消费者的消费速度跟不上生产者的生产速度时，就会导致消息堆积在队列中，而默认消息是没有持久化的，存在于内存之中，所以假如服务器宕机等故障发生，就会导致队列中的数据丢失。

这里的解决方案也很简单，就是将消息进行持久化，在 rabbitmq 当中，持久化也可以分为 3 种：交换机持久化，队列持久化和消息持久化。

虽然说持久化能一定程度上保证消息的可靠性，然而当出现了服务器的磁盘损坏，依然可能出现消息丢失，所以为了更加完美，rabbitmq 集群可能是必须的，当然，本文不会涉及到集群的知识，集群的知识以及搭建会放到下次再来分析。

交换机持久化

声明交换机时，durable 参数设置为 true。

队列持久化

声明队列时，durable 参数设置为 true。

消息持久化

发送消息时可以将消息设置为持久化。

java api 消息持久化

在 java api 中，可以通过如下方式设置消息持久化：

//deliverymode=2表示消息持久化

amqp.basicproperties properties = new amqp.basicproperties.builder().deliverymode(2).build();

channel.basicpublish(“exchangename”,“routingkey”,properties,msg.getbytes());

spring boot 消息持久化

在 spring boot 中可以通过如下方式将消息设置为持久化：

messageproperties messageproperties = new messageproperties();

messageproperties.setdeliverymode(messagedeliverymode.persistent);//消息持久化

message message = new message(msg.getbytes(), messageproperties);

rabbittemplate.convertandsend(“exchangename”,“routingkey”,message);

消费者消费消息失败了怎么办

---

踏遍千山万水，经过 3 层地狱模式，消息终于被消费者拿到手了，然而悲剧的事情又发生了，消费者消费消息的时候可能因为消费者本身的问题或者其他意外导致了消费者消费消息失败了，这时候消息还是没能被正确处理，这时候难道眼睁睁看着最后关头了束手无策了吗？

非也，作为一款如此优秀的消息队列，怎么可能没考虑到这种场景呢。还记不记得上面我们提到的确认模式，实际上，上面的两种确认模式都属于服务端的确认，在 rabbitmq 中为消费者也提供了确认模式，这就是消费者的确认。

消费者确认（ack）

队列当中会把消息删除的前提就是这条消息被消费者消费掉了，但是服务器如何知道消息被消费了呢？这就是需要通过消费者确认之后才会删除，而我们前面在介绍消息发送的时候貌似并没有看到消费者确认流程，这是因为消费者默认在收到消息后会给服务器一个应答，服务端收到消费者的应答之后，就会删除消息。

java api 实现消费者应答

在 java api 中应答方式有两种，自动应答和手动应答，当自动应答时，则只要消费者收到消息就会给服务端确认，不在乎消息是否消费成功。

• 1、新建一个消费者 ackconsumer 类（省略了包名和导入），这里为了实现方便，通过生产者的头部标记来决定采用何种应答策略：

public class ackconsumer {

private static string queue_name = “ack_queue”;

public static void main(string[] args) throws exception{

//1.声明连接

connectionfactory factory = new connectionfactory();

factory.seturi(“amqp://username:password@ip:port”);

//2.建立连接

connection conn = factory.newconnection();

//3.创建消息通道

channel channel = conn.createchannel();

//4.声明队列（默认交换机amqp default，direct）

channel.queuedeclare(queue_name, false, false, false, null);

system.out.println(" 等待接收消息…");

// 创建消费者

consumer consumer = new defaultconsumer(channel) {

@override

public void handledelivery(string consumertag, envelope envelope, amqp.basicproperties properties,

byte[] body) throws ioexception {

system.out.println("收到消息: " + new string(body, “utf-8”));

map<string,object> map = properties.getheaders();//获取头部消息

string acktype = map.get(“acktype”).tostring();

if (acktype.equals(“ack”)){//手动应答

channel.basicack(envelope.getdeliverytag(),true);

}else if(acktype.equals(“reject-single”)){//拒绝单条消息

//拒绝消息。requeue参数表示消息是否重新入队

channel.basicreject(envelope.getdeliverytag(),false);

// channel.basicnack(envelope.getdeliverytag(),false,false);

}else if (acktype.equals(“reject-multiple”)){//拒绝多条消息

//拒绝消息。multiple参数表示是否批量拒绝，为true则表示<deliverytag的消息都被拒绝

channel.basicnack(envelope.getdeliverytag(),true,false);

}

}

};

//开始获取消息,第二个参数 autoack表示是否开启自动应答

channel.basicconsume(queue_name, false, consumer);

}

}

• 2、新建一个生产者 ackproducer 类（省略了包名和导入）：

public class ackproducter {

private static string queue_name = “ack_queue”;//队列

private static string exchange_name = “ack_exchange”;//交换机

private static string routeing_key = “test”;

public static void main(string[] args) throws exception {

connectionfactory factory = new connectionfactory();

factory.seturi(“amqp://admin:123456@47.107.155.197:5672”);

// 建立连接

connection conn = factory.newconnection();

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、oppo等大厂，18年进入阿里一直到现在。

深知大多数java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上java开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！

如果你觉得这些内容对你有帮助，可以扫码获取！！（备注java获取）

总结

大型分布式系统犹如一个生命，系统中各个服务犹如骨骼，其中的数据犹如血液，而kafka犹如经络，串联整个系统。这份kafka源码笔记通过大量的设计图展示、代码分析、示例分享，把kafka的实现脉络展示在读者面前，帮助读者更好地研读kafka代码。

麻烦帮忙转发一下这篇文章+关注我

《互联网大厂面试真题解析、进阶开发核心学习笔记、全套讲解视频、实战项目源码讲义》
queue";//队列

private static string exchange_name = “ack_exchange”;//交换机

private static string routeing_key = “test”;

public static void main(string[] args) throws exception {

connectionfactory factory = new connectionfactory();

factory.seturi(“amqp://admin:123456@47.107.155.197:5672”);

// 建立连接

connection conn = factory.newconnection();

自我介绍一下，小编13年上海交大毕业，曾经在小公司待过，也去过华为、oppo等大厂，18年进入阿里一直到现在。

深知大多数java工程师，想要提升技能，往往是自己摸索成长或者是报班学习，但对于培训机构动则几千的学费，着实压力不小。自己不成体系的自学效果低效又漫长，而且极易碰到天花板技术停滞不前！

因此收集整理了一份《2024年java开发全套学习资料》，初衷也很简单，就是希望能够帮助到想自学提升又不知道该从何学起的朋友，同时减轻大家的负担。[外链图片转存中…(img-cb12koys-1713373104168)]

[外链图片转存中…(img-pkyqzdfs-1713373104168)]

[外链图片转存中…(img-5tegy1nd-1713373104169)]

既有适合小白学习的零基础资料，也有适合3年以上经验的小伙伴深入学习提升的进阶课程，基本涵盖了95%以上java开发知识点，真正体系化！

由于文件比较大，这里只是将部分目录截图出来，每个节点里面都包含大厂面经、学习笔记、源码讲义、实战项目、讲解视频，并且会持续更新！

如果你觉得这些内容对你有帮助，可以扫码获取！！（备注java获取）

总结

大型分布式系统犹如一个生命，系统中各个服务犹如骨骼，其中的数据犹如血液，而kafka犹如经络，串联整个系统。这份kafka源码笔记通过大量的设计图展示、代码分析、示例分享，把kafka的实现脉络展示在读者面前，帮助读者更好地研读kafka代码。

麻烦帮忙转发一下这篇文章+关注我

[外链图片转存中…(img-n69vsirn-1713373104169)]

《互联网大厂面试真题解析、进阶开发核心学习笔记、全套讲解视频、实战项目源码讲义》