rabbitMQ的高级特性01
作者：互联网
1.消息的可靠性传递
在使用 RabbitMQ 的时候，作为消息发送方希望杜绝任何消息丢失或者投递失败场景。
RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式。
（1）确认模式
即消息需要经过服务器端确认后，才会消费掉
开启确认模式
在spingboot项目中resources文件夹内创建一个.yml文件
修改.yml文件中的内容为：
spring:
rabbitmq:
host: 192.168.192.129
#开启确认模式
publisher-confirm-type: correlated
设置RabbitTemplate的回调函数
@Autowired
private RabbitTemplate rabbitTemplate;
@Test
public void testConfirm(){
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
if(b==false){//消息没有到达交换机 根据业务需求。
System.out.println("继续发送消息");
}
}
});
rabbitTemplate.convertAndSend("ban129_exchange","","hello confirm");
}
（2）回退模式
即消息如果发送到队列失败，则会回退到交换机中，不会直接丢失消息
开启回退模式
spring:
rabbitmq:
host: 192.168.192.129
#开启确认模式
publisher-confirm-type: correlated
#开启回退模式
publisher-returns: true
设置RabbitTemplate的回调函数
public void testReturn(){
rabbitTemplate.setReturnsCallback(new RabbitTemplate.ReturnsCallback() {
@Override
public void returnedMessage(ReturnedMessage returnedMessage) {
//只要交换机到队列失败时才会触发该方法。 可以继续发送也可以取消相应的业务功能。
System.out.println("消息从交换机到队列失败"+returnedMessage.getReplyText());
}
});
rabbitTemplate.convertAndSend("ban129_exchange_direct","error2","hello confirm2");
}
2.Consumer ACK
前两种方式都是保证了生产者的消息传递的可靠性，而针对于消费者消息传递的可靠性，可以靠Consumer ACK来实现
（1）消费者配置手动开启确认模式
spring:
rabbitmq:
host: 192.168.192.129
listener:
direct:
#开启手动确认模式 manual 自动确认模式 none
acknowledge-mode: manual
（2）测试
package com.wyj;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import java.io.IOException;
/**
* @author 王逸君
* @version 1.0
* @date 2021/4/21 21:39
*/
public class Test {
@Autowired
private RabbitTemplate rabbitTemplate;
@RabbitListener(queues = "qy129_direct01")
public void listener(Message message, Channel channel) throws IOException {
long tag = message.getMessageProperties().getDeliveryTag();
//获取消息，并且转换为字符串
byte[] body = message.getBody();
String msg=new String(body);
/**
* long deliveryTag,
* boolean multiple 是否允许多段确认
*/
try {
//如果接收消息处理逻辑时出现异常
int a=10/0;
System.out.println("处理业务逻辑");
channel.basicAck(tag,true);//从队列中删除消息
} catch (Exception e) {
//捕获异常后,让队列再次发送消息
/**
* long deliveryTag,
* boolean multiple,是否允许多段确认
* boolean requeue 是否再次发送消息
*/
channel.basicNack(tag,true,true);
}
AMQP.BasicProperties builder = new AMQP.BasicProperties.Builder().expiration("60000").build();
}
}
3.消费端限流
（1）开启手动确认模式以及设置限流条数
spring:
rabbitmq:
host: 192.168.192.129
listener:
direct:
#开启手动确认模式 manual 自动确认模式 none
acknowledge-mode: manual
#消息段限流，每次处理10条消息
prefetch: 10
（2）测试
package com.wyj;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import java.io.IOException;
/**
* @author 王逸君
* @version 1.0
* @date 2021/4/21 21:39
*/
public class Test {
@Autowired
private RabbitTemplate rabbitTemplate;
@RabbitListener(queues = "qy129_direct01")
public void listener(Message message, Channel channel) throws IOException {
long tag = message.getMessageProperties().getDeliveryTag();
//获取消息，并且转换为字符串
byte[] body = message.getBody();
String msg=new String(body);
/**
* long deliveryTag,
* boolean multiple 是否允许多段确认
*/
try {
//如果接收消息处理逻辑时出现异常
int a=10/0;
System.out.println("处理业务逻辑");
channel.basicAck(tag,true);//从队列中删除消息
} catch (Exception e) {
//捕获异常后,让队列再次发送消息
/**
* long deliveryTag,
* boolean multiple,是否允许多段确认
* boolean requeue 是否再次发送消息
*/
channel.basicNack(tag,true,true);
}
AMQP.BasicProperties builder = new AMQP.BasicProperties.Builder().expiration("60000").build();
}
}
4.TTL队列/消息
TTL是Time To Live的缩写, 也就是生存时间
RabbitMQ支持消息的过期时间, 在消息发送时可以进行指定
RabbitMQ支持队列的过期时间, 从消息入队列开始计算, 只要超过了队列的超时时间配置, 那么消息会自动清除
代码实现如下：
@Test
public void testSend(){
rabbitTemplate.convertAndSend("exchange01","","wyj确实帅");
}
@Test
public void testSend02(){
for(int i=0;i<10;i++) {
if(i==3){
MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
@Override
public Message postProcessMessage(Message message) throws AmqpException {
message.getMessageProperties().setExpiration("5000");
return message;
}
};
rabbitTemplate.convertAndSend("exchange01", "", "wyj确实帅"+i, messagePostProcessor);
}else {
rabbitTemplate.convertAndSend("exchange01", "", "wyj确实帅"+i);
}
}
}
标签：rabbitTemplate,RabbitTemplate,01,特性,rabbitMQ,import,message,public,消息 来源： https://blog.csdn.net/ttcxyx/article/details/116241952