RabbitMq学习——Springboot整合rabbitmq之手动消息确认(ACK)_springboot rabbitmq 手动ack

一、前言

前几天我研究了关于springboot整合简单消息队列，实现springboot推送消息至队列中，消费者成功消费。同时也加了消息转发器，对消息转发器各种类型的配置等做了总结。
但是，主要还有一点，我一直存在疑问：如何确保消息成功被消费者消费？

---

说到这里，我相信很多人会说使用ack啊，关闭队列自动删除啊什么的。主要是道理大家都懂，我要实际的代码，网上找了半天，和我设想的有很大差异，还是自己做研究总结吧。

二、准备

本次写案例，就按照最简单的方式，direct方式进行配置吧，实际流程如下所示：

1、消息转发器类型：direct直连方式。
2、消息队列：暂时采取公平分发方式。
3、实现流程：消息生产者生产的消息发送至队列中，由两个消费者获取并消费，消费完成后，清楚消息队列中的消息。

所以我们接下来先写配置和demo。

2.1、依赖引入

再一般的springboot 2.1.4项目中，添加一个pom依赖。

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
1
2
3
4

2.2、连接yml的配置

我们这边暂时只有一个rabbitmq，所以连接操作，基本rabbitmq的信息配置问题直接再yml中编写就可以了。

spring:
  rabbitmq:
    host: 127.0.0.1
    port: 5672
    username: xiangjiao
    password: bunana
    virtual-host: /xiangjiao
    publisher-confirms: true   #开启发送确认
    publisher-returns: true  #开启发送失败回退
    
    #开启ack
    listener:
      direct:
        acknowledge-mode: manual
      simple:
        acknowledge-mode: manual #采取手动应答
        #concurrency: 1 # 指定最小的消费者数量
        #max-concurrency: 1 #指定最大的消费者数量
        retry:
          enabled: true # 是否支持重试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

2.3、config注入配置

我们根据图示

知道我们必须配置以下东西：
1、一个消息转发器，我们取名directExchangeTx。
2、一个消息队列，取名directQueueTx，并将其绑定至指定的消息转发器上。

所以我们的配置文件需要这么写：

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 直连交换机，发送指定队列信息，但这个队列后有两个消费者同时进行消费
 * @author 7651
 *
 */
@Configuration
public class DirectExchangeTxQueueConfig {
	
	@Bean(name="getDirectExchangeTx")
	public DirectExchange getDirectExchangeTx(){
		return new DirectExchange("directExchangeTx", true, false);
	}
	
	@Bean(name="getQueueTx")
	public Queue getQueueTx(){
		return new Queue("directQueueTx", true, false, false);
	}
	
	@Bean
	public Binding getDirectExchangeQueueTx(
			@Qualifier(value="getDirectExchangeTx") DirectExchange getDirectExchangeTx,
			@Qualifier(value="getQueueTx") Queue getQueueTx){
		return BindingBuilder.bind(getQueueTx).to(getDirectExchangeTx).with("directQueueTxRoutingKey");
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

2.4、消费者的配置

有了队列和消息转发器，消息当然需要去消费啊，所以我们接下来配置消息消费者。

从图中，我们看出，我们需要配置两个消息消费者，同时监听一个队列，所以我们的配置类为：
消费者一：

import java.io.IOException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;

@Component
@RabbitListener(queues="directQueueTx")
public class Consumer1 {
	@RabbitHandler
	public void process(String msg,Channel channel, Message message) throws IOException {
		//拿到消息延迟消费
		try {
			Thread.sleep(1000*1);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		
		try {
			/**
			 * 确认一条消息：<br>
			 * channel.basicAck(deliveryTag, false); <br>
			 * deliveryTag:该消息的index <br>
			 * multiple：是否批量.true:将一次性ack所有小于deliveryTag的消息 <br>
			 */
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
			System.out.println("get msg1 success msg = "+msg);
			
		} catch (Exception e) {
			//消费者处理出了问题，需要告诉队列信息消费失败
			/**
			 * 拒绝确认消息:<br>
			 * channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ; <br>
			 * deliveryTag:该消息的index<br>
			 * multiple：是否批量.true:将一次性拒绝所有小于deliveryTag的消息。<br>
			 * requeue：被拒绝的是否重新入队列 <br>
			 */
			channel.basicNack(message.getMessageProperties().getDeliveryTag(),
					false, true);
			System.err.println("get msg1 failed msg = "+msg);
			
			/**
			 * 拒绝一条消息：<br>
			 * channel.basicReject(long deliveryTag, boolean requeue);<br>
			 * deliveryTag:该消息的index<br>
			 * requeue：被拒绝的是否重新入队列 
			 */
			//channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53

消息消费者二：

import java.io.IOException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;

@Component
@RabbitListener(queues="directQueueTx")
public class Consumer2 {
	@RabbitHandler
	public void process(String msg,Channel channel, Message message) throws IOException {
		//拿到消息延迟消费
		try {
			Thread.sleep(1000*3);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		
		try {
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
			System.out.println("get msg2 success msg = "+msg);
			
		} catch (Exception e) {
			//消费者处理出了问题，需要告诉队列信息消费失败
			channel.basicNack(message.getMessageProperties().getDeliveryTag(),
					false, true);
			System.err.println("get msg2 failed msg = "+msg);
		}
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32

两个消费者之间唯一的区别在于两者获取消息后，延迟时间不一致。

2.5、消息生产者

有了消息消费者，我们需要有一个方式提供消息并将消息推送到消息队列中。

public interface IMessageServcie {
	public void sendMessage(String exchange,String routingKey,Object msg);
}
1
2
3

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ConfirmCallback;
import org.springframework.amqp.rabbit.core.RabbitTemplate.ReturnCallback;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import cn.linkpower.service.IMessageServcie;

@Component
public class MessageServiceImpl implements IMessageServcie,ConfirmCallback,ReturnCallback {
	
	private static Logger log = LoggerFactory.getLogger(MessageServiceImpl.class);
	
	@Autowired
	private RabbitTemplate rabbitTemplate;
	
	@Override
	public void sendMessage(String exchange,String routingKey,Object msg) {
		//消息发送失败返回到队列中, yml需要配置 publisher-returns: true
		rabbitTemplate.setMandatory(true);
		//消息消费者确认收到消息后，手动ack回执
		rabbitTemplate.setConfirmCallback(this);
		rabbitTemplate.setReturnCallback(this);
		//发送消息
		rabbitTemplate.convertAndSend(exchange,routingKey,msg);
	}

	@Override
	public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
		log.info("---- returnedMessage ----replyCode="+replyCode+" replyText="+replyText+" ");
	}
	
	@Override
	public void confirm(CorrelationData correlationData, boolean ack, String cause) {
		log.info("---- confirm ----ack="+ack+"  cause="+String.valueOf(cause));
		log.info("correlationData -->"+correlationData.toString());
		if(ack){
			log.info("---- confirm ----ack==true  cause="+cause);
		}else{
			log.info("---- confirm ----ack==false  cause="+cause);
		}
	}

}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47

除了定义好了消息发送的工具服务接口外，我们还需要一个类，实现请求时产生消息，所以我们写一个controller。

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.ResponseBody;
import cn.linkpower.service.IMessageServcie;

@Controller
public class SendMessageTx {
	
	@Autowired
	private IMessageServcie messageServiceImpl;
	
	@RequestMapping("/sendMoreMsgTx")
	@ResponseBody
	public String sendMoreMsgTx(){
		//发送10条消息
		for (int i = 0; i < 10; i++) {
			String msg = "msg"+i;
			System.out.println("发送消息  msg："+msg);
			messageServiceImpl.sendMessage("directExchangeTx", "directQueueTxRoutingKey", msg);
			//每两秒发送一次
			try {
				Thread.sleep(2000);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
		return "send ok";
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

运行springboot项目，访问指定的url，是可以观察到消息产生和消费的。

有些人会问，写到这里就够了吗，你这和之前博客相比，和没写一样啊，都是教我们如何配置，如何生产消息，如何消费消息。

所以接下来的才是重点了，我们一起研究一个事，当我们配置的消费者二出现消费消息时，出问题了，你如何能够保证像之前那样，消费者一处理剩下的消息？

三、ack配置和测试

3.1、模拟消费者二出问题

我们发送的消息格式都是 msg1、msg2、…

所以，我们不妨这么想，当我消费者二拿到的消息msg后面的数字大于3，表示我不要了。

import java.io.IOException;

import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import com.rabbitmq.client.Channel;

@Component
@RabbitListener(queues="directQueueTx")
public class Consumer2 {
	@RabbitHandler
	public void process(String msg,Channel channel, Message message) throws IOException {
		//拿到消息延迟消费
		try {
			Thread.sleep(1000*3);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		
		
		try {
			if(!isNull(msg)){
				String numstr = msg.substring(3);
				Integer num = Integer.parseInt(numstr);
				if(num >= 3){
					channel.basicNack(message.getMessageProperties().getDeliveryTag(),
							false, true);
					System.out.println("get msg2 basicNack msg = "+msg);
				}else{
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
					System.out.println("get msg2 basicAck msg = "+msg);
				}
			}
		} catch (Exception e) {
			//消费者处理出了问题，需要告诉队列信息消费失败
			channel.basicNack(message.getMessageProperties().getDeliveryTag(),
					false, true);
			System.err.println("get msg2 failed msg = "+msg);
		}
	}
	
	public static boolean isNull(Object obj){
		return obj == null || obj == ""||obj == "null";
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

再次请求接口，我们统计日志信息打印发现：

发现：

当我们对消息者二进行限制大于等于3时，不接受消息队列传递来的消息时，消息队列会随机重发那条消息，直至消息发送至完好的消费者一时，才会把消息消费掉。

四、分析几个回执方法

4.1、确认消息

channel.basicAck(long deliveryTag, boolean multiple);

我们一般使用下列方式：

channel.basicAck(
message.getMessageProperties().getDeliveryTag(), 
false);
1
2
3

4.2、拒绝消息

channel.basicNack(long deliveryTag, boolean multiple, boolean requeue) ;

我们接下来还是修改消费者二，将这个方法最后个参数更改为false，看现象是什么？

import java.io.IOException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Channel;

@Component
@RabbitListener(queues="directQueueTx")
public class Consumer2 {
	@RabbitHandler
	public void process(String msg,Channel channel, Message message) throws IOException {
		//拿到消息延迟消费
		try {
			Thread.sleep(1000*3);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
		
		
		
		try {
			if(!isNull(msg)){
				String numstr = msg.substring(3);
				Integer num = Integer.parseInt(numstr);
				if(num >= 3){
					channel.basicNack(message.getMessageProperties().getDeliveryTag(),
							false, false);
					System.out.println("get msg2 basicNack msg = "+msg);
				}else{
					channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
					System.out.println("get msg2 basicAck msg = "+msg);
				}
			}
		} catch (Exception e) {
			//消费者处理出了问题，需要告诉队列信息消费失败
			channel.basicNack(message.getMessageProperties().getDeliveryTag(),
					false, true);
			System.err.println("get msg2 failed msg = "+msg);
		}
	}
	
	public static boolean isNull(Object obj){
		return obj == null || obj == ""||obj == "null";
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

重启项目，重新请求测试接口。

发现，当出现设置参数为false时，也就是如下所示的设置时：

channel.basicNack(
	message.getMessageProperties().getDeliveryTag(),
	false, 
	false);
1
2
3
4

如果此时消费者二出了问题，这条消息不会重新回归队列中重新发送，会丢失这条数据。
并且再消息队列中不会保存：

4.3、拒绝消息

channel.basicReject(long deliveryTag, boolean requeue);

这个和上面的channel.basicNack又有什么不同呢？我们还是修改消费者二实验下。

请求测试接口，查看日志信息。

发现，此时的日志信息配置

channel.basicReject(
message.getMessageProperties().getDeliveryTag(),
 true);
1
2
3

和

channel.basicNack(
message.getMessageProperties().getDeliveryTag(),
false, true);
1
2
3

实现的效果是一样的，都是将信息拒绝接收，由于设置的requeue为true，所以都会将拒绝的消息重新入队列中，重新进行消息分配并消费。

五、总结

这一篇博客，我们总结了相关的配置，三个确认（或回执）信息的方法，并区别了他们的各项属性，也知道了当消息再一个消费者中处理失败了，如何不丢失消息重新进行消息的分配消费问题。

但是这个只是队列和消费者之间的消息确认机制，使用手动ACK方式确保消息队列中的消息都能在消费者中成功消费。那么，消息转发器和消息队列之间呢？消息生产者和消息转发器之间呢？

这些问题我们留在下一篇博客中再分别讨论吧。

当然，差点忘了一个小问题。
我们思考一个问题，如果消息队列对应的消费者只有一个，并且那个消费者炸了，会出现什么问题呢？？