rabbitmq消息可靠之手动确认

rabbitmq消息可靠之手动确认

rabbitmq 消息值之手动确认

Consume

新建springboot项目，导入rabbitmq依赖

  <dependency>           <groupId>org.springframework.boot</groupId>           <artifactId>spring-boot-starter-amqp</artifactId>  </dependency>

修改配置文件

server:port: 8102  #服务端口
​
spring:rabbitmq:   host:  rabbitmqhost    #rabbitmq地址   port: 15672   #rabbitmq amqp端口   username: root  #rabbitmq用户名   password: root  #rabbitmq密码   virtual-host: serve #rabbitmq虚拟主机   listener:     type: simple     simple:       acknowledge-mode: MANUAL # MANUAL消息确认方式 手动确认 ，none 不确认 ，auto自动确认       retry:         enabled: true  #开启重试         max-attempts: 3 #最大重试次数         initial-interval: 5000ms #重试间隔时间

启动类增加@EnableRabbit

ample.demo;
​
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
​
​
@SpringBootApplication
@EnableRabbit
public class DemoApplication {
​public static void main(String[] args) {
​SpringApplication.run(DemoApplication.class, args);}
​
}

新增监听类MsgListener

ample.demo;
​
/*** @Author: lyh* @Date: 2021/12/14 20:47*/
​
/*** @Author: lyh* @Date: 2021/12/13 17:21*/
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import org.Message;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;
​
import java.io.IOException;
​
/*** @Author: lyh* @Date: 2021/12/13 14:49*/
@Component
public class MsgListen {   @RabbitListener(           bindings = {@QueueBinding(value = @Queue(value = "blade-user-queue",durable = "ture"),                   exchange = @Exchange(value = "serve-exchange"),key="blade-user")})   public void consumer(Channel channel, Message message)throws IOException{       System.out.println("收到消息"+MessageProperties());       channel.MessageProperties().getDeliveryTag(),true);   }
}
​

消费者端进行绑定消息队列，非则生产者发送消息会产生路由消息失败问题

Produce

生产者配置

由于生产者可能会遇到消息发送失败的情况，所以需要用数据库进行持久化，来保证消息的可靠性

#数据源配置
spring:rabbitmq:   host: 36.103.242.17   port: 25672   username: bladex   password: bladex   virtual-host: serve   retry-interval: 10s   exchange: serve-exchange   max-retry-times: 3