1.引入 spring-boot-starter-amqp
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
2. application.properties配置
# RabbitMQ配置
spring.rabbitmq.host=192.168.56.10
spring.rabbitmq.port=5672
# 虚拟主机配置**********这些是消息抵达配置****
spring.rabbitmq.virtual-host=/
# 开启发送端消息抵达Broker确认
spring.rabbitmq.publisher-confirms=true
# 开启发送端消息抵达Queue确认
spring.rabbitmq.publisher-returns=true
# 只要消息抵达Queue,就会异步发送优先回调returnfirm
spring.rabbitmq.template.mandatory=true
# 手动ack消息,不使用默认的消费端确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
3.确认抵达
- 消费者获取到消息,成功处理,可以回复Ack给Broker
basic.ack用于肯定确认;broker将移除此消息
basic.nack用于否定确认;可以指定broker是否丢弃此消息,可以批量
basic.reject用于否定确认;同上,但不能批量
默认自动ack,消息被消费者收到,就会从broker的queue中移除
queue无消费者,消息依然会被存储,直到消费者消费
消费者收到消息,默认会自动ack。但是如果无法确定此消息是否被处理完成, 或者成功处理。我们可以开启手动ack模式
消息处理成功,ack(),接受下一个消息,此消息broker就会移除
消息处理失败,nack()/reject(),重新发送给其他人进行处理,或者容错处理后ack
消息一直没有调用ack/nack方法,broker认为此消息正在被处理,不会投递给别人,此时客户 端断开,消息不会被broker移除,会投递给别人
4.实现
package com.xunqi.gulimall.order.listener;
import com.rabbitmq.client.Channel;
import com.xunqi.gulimall.order.entity.OrderEntity;
import com.xunqi.gulimall.order.service.OrderService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitHandler;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.io.IOException;
/**
* @Description: 定时关闭订单
* @Created: with IntelliJ IDEA.
* @author: LY
* @createTime: 2020-07-07 09:54
**/
@RabbitListener(queues = "order.release.order.queue")
@Service
public class OrderCloseListener {
@Autowired
private OrderService orderService;
@RabbitHandler// 可以根据参数不同具有重载效果
public void listener(OrderEntity orderEntity, Channel channel, Message message) throws IOException {
System.out.println("收到过期的订单信息,准备关闭订单" + orderEntity.getOrderSn());
try {
orderService.closeOrder(orderEntity);
//message.getMessageProperties().getDeliveryTag()
// channel 内按顺序自增的消息唯一标识号; false 不做批量签收
//手动确认模式 basicAck ;
channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);
} catch (Exception e) {
//网络中断了 basicReject(long deliveryTag, boolean requeue)了拒签
// 拒签:basicNack(long deliveryTag, boolean multiple, boolean requeue: 如果为 false 丢弃,ture 发回消息队列,服务器重新入队)
channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);
}
}
}