RabbitMQ开启手动签收模式配置
rabbitmq:
####连接地址
host: 127.0.0.1
####端口号
port: 5672
####账号
username: guest
####密码
password: guest
### 地址
virtual-host: /
listener:
simple:
retry:
####开启消费者(程序出现异常的情况下会)进行重试
enabled: true
####最大重试次数
max-attempts: 5
####重试间隔次数
initial-interval: 3000
####开启手动ack
acknowledge-mode: manual
接收消息
@RabbitListener(queues = "fanout_email_queue")
public void process(Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
String messageId = message.getMessageProperties().getMessageId();
String msg = new String(message.getBody(), "UTF-8");
System.out.println("邮件消费者获取生产者消息msg:" + msg + ",消息id:" + messageId);
// 重试机制都是间隔性
JSONObject jsonObject = JSONObject.parseObject(msg);
String email = jsonObject.getString("email");
String emailUrl = "http://127.0.0.1:8083/sendEmail?email=" + email;
System.out.println("邮件消费者开始调用第三方邮件服务器,emailUrl:" + emailUrl);
JSONObject result = HttpClientUtils.httpGet(emailUrl);
// 如果调用第三方邮件接口无法访问,如何实现自动重试.
if (result == null) {
throw new Exception("调用第三方邮件服务器接口失败!");
}
System.out.println("邮件消费者结束调用第三方邮件服务器成功,result:" + result + "程序执行结束");
// 手动ack
Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
// 手动签收
channel.basicAck(deliveryTag, false);
// channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
}
// 默认是自动应答模式