rabbitmq高级特性（消息手动确认）

为了保证消息不丢失，从生产者两种模式（确认模式和返回模式）到消费者手动确认的一个大整合

生产者

1.maven依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind --><!-- 用于rabbitmq发送json消息 --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.7.5</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency>

2.配置文件yml

server:port: 8001spring:rabbitmq:host: 192.168.56.101port: 5672virtual-host: /emsusername: adminpassword: admin# 开启确认模式publisher-confirms: true# 开启回退模式publisher-returns: true

3.编写rabbitmq配置类RabbitmqConfig，包含交换机、队列、绑定队列与交换机、设置确认模式和返回模式以及消息转换器

/*** @author xiaozikang* @date 2020/10/10 11:02* @Email:xiaozikangwy@163.com*/@Configuration
public class RabbitmqConfig {public static final String CONFIRM_EXCAHNGE = "confirmExchange";public static final String CONFIRM_QUEUE = "confirmQueue";// 交换机@Bean("confirmExchange")public Exchange confirmExchange(){return ExchangeBuilder.directExchange(CONFIRM_EXCAHNGE).durable(true).build();}// 队列@Bean("confirmQueue")public Queue confirmQueue(){return QueueBuilder.durable(CONFIRM_QUEUE).build();}// 绑定队列与交换机@Beanpublic Binding binding(@Qualifier("confirmQueue") Queue confirmQueue,@Qualifier("confirmExchange") Exchange confirmExchange){return BindingBuilder.bind(confirmQueue).to(confirmExchange).with("confirm").noargs();}@Beanpublic RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);// 定义消息转换器Jackson2JsonMessageConverter，发送json数据rabbitTemplate.setMessageConverter(producerJackson2MessageConverter());// 设置开启Mandatory，如果消息没路由到队列，则返回给消息发送方（如果不开启，默认丢弃消息）rabbitTemplate.setMandatory(true);// 定义确认模式rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {/***  correlationData 相关配置信息*  ack exchange交换机是否成功收到了消息，true成功*  cause 失败原因*/if (ack){System.out.println("exchange成功收到消息");}else {System.out.print("未收到消息：" + cause);// 可以做些处理，再次发送}}});// 定义返回模式rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {@Overridepublic void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {/***  message 消息对象*  replyCode 错误码*  replyText 错误信息*  exchange 交换机*  routingKey 路由键*/System.out.println("回调函数执行了");}});return rabbitTemplate;}@Beanpublic Jackson2JsonMessageConverter producerJackson2MessageConverter(){return new Jackson2JsonMessageConverter();}}

4.实体类（主要用于测试发送json消息）

@Data
@NoArgsConstructor
@AllArgsConstructor
public class User implements Serializable {private String name;private Integer age;
}

5.发送消息

@Testpublic void testf01(){User user = new User("小明",10);for (int i = 0; i < 5; i++) {rabbitTemplate.convertAndSend(RabbitmqConfig.CONFIRM_EXCAHNGE,"confirm",user);}}

消费者

1.maven依赖

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-amqp</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><optional>true</optional></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope><exclusions><exclusion><groupId>org.junit.vintage</groupId><artifactId>junit-vintage-engine</artifactId></exclusion></exclusions></dependency><dependency><groupId>org.springframework.amqp</groupId><artifactId>spring-rabbit-test</artifactId><scope>test</scope></dependency><!-- https://mvnrepository.com/artifact/com.fasterxml.jackson.core/jackson-databind --><dependency><groupId>com.fasterxml.jackson.core</groupId><artifactId>jackson-databind</artifactId><version>2.7.5</version></dependency>

2.配置文件yml

server:port: 9020spring:rabbitmq:host: 192.168.56.101port: 5672virtual-host: /emsusername: adminpassword: adminlistener:simple:# 消费手动确认acknowledge-mode: manual# 消费者每次从队列获取的消息数量。此属性当不设置时为：轮询分发，设置为1为：公平分发prefetch: 2# 指定最小的消费者数量# concurrency: 1# 指定最大的消费者数量# max-concurrency: 2

3.编写配置类（这边主要用于接受json类型消息）

/*** @author xiaozikang* @date 2020/10/12 17:50* @Email:xiaozikangwy@163.com*/
@Configuration
public class RabbitConfig implements RabbitListenerConfigurer {@Overridepublic void configureRabbitListeners(RabbitListenerEndpointRegistrar registrar) {registrar.setMessageHandlerMethodFactory(messageHandlerMethodFactory());}@BeanMessageHandlerMethodFactory messageHandlerMethodFactory() {DefaultMessageHandlerMethodFactory messageHandlerMethodFactory = new DefaultMessageHandlerMethodFactory();messageHandlerMethodFactory.setMessageConverter(consumerJackson2MessageConverter());return messageHandlerMethodFactory;}@Beanpublic MappingJackson2MessageConverter consumerJackson2MessageConverter() {return new MappingJackson2MessageConverter();}
}

4.消费消息

@Component
public class Consumer {@RabbitListener(queues = "confirmQueue")public void process(User user, Channel channel, Message message) throws IOException {// 1.接受转换消息long deliveryTag = message.getMessageProperties().getDeliveryTag();try {// 2.处理业务逻辑System.out.println("处理业务逻辑。。。。。。");// 3.手动签收// channel.basicAck(deliveryTag,true);} catch (Exception e) {// e.printStackTrace();// 4.出现异常，拒绝签收/*** requeue 重回队列* multiple false只确认当前consumer一个消息收到，true确认所有consumer获得的消息*/channel.basicNack(deliveryTag,true,true);}}}

rabbitmq高级特性（消息手动确认）相关推荐

RabbitMQ 高级特性（吐血猝死整理篇）
文章目录 RabbitMQ 高级特性消息可靠性投递(可靠性发送) 事务机制代码实现发送方确认机制为什么比事务性能好示例代码测试一下QPS 持久化存储 TTL 队列死信队列(DLX) 延迟 ...
RabbitMQ（二）：RabbitMQ高级特性
RabbitMQ(二):RabbitMQ高级特性 RabbitMQ是目前非常热门的一款消息中间件,不管是互联网大厂还是中小企业都在大量使用.作为一名合格的开发者,有必要了解一下相关知识,RabbitM ...
RabbitMQ高级特性
文章目录 1. 简述 2. 特性示例: 2.1 消息可靠性投递 2.2 Consumer Ack 2.3 消费端限流 2.4 TTL 2.5 死信队列 2.6 延迟队列 1. 简述在rabbitMQ ...
3 RabbitMQ高级特性 3
主要为大家讲解RabbitMQ的高级特性和实际场景应用, 包括消息如何保障 100% 的投递成功 ? 幂等性概念详解,在海量订单产生的业务高峰期,如何避免消息的重复消费问题? Confirm确认消息. ...
【消息中间件】RabbitMQ 高级特性与应用问题
消息的可靠投递在使用 RabbitMQ 的时候,作为消息发送方希望杜绝任何消息丢失或者投递失败场景.RabbitMQ 为我们提供了两种方式用来控制消息的投递可靠性模式. confirm 确认模式 r ...
1-8 (4). RabbitMQ高级特性-消费端ACK
Consumer ACK 指Acknowledge,确认有三种方式: (1)自动确认:acknowledge="none"(默认) (2)手动确认:acknowledge=&qu ...
RabbitMQ高级特性——死信队列DLX以及代码测试
大伙可以到我的RabbitMQ专栏获取更多信息 demo示例这里拿概述死信队列,缩写DLX(dead letter exchange 死信交换机),当消息称为dead message之后,会被重新 ...
RabbitMQ高级特性（五）：RabbitMQ之死信队列DLX
一.死信队列简介 (1)死信队列死信队列,英文缩写:DLX .Dead Letter Exchange(死信交换机),当消息成为Dead message后,可以被重新发送到另一个交换机,这个交换机就 ...
RabbitMQ（13）RabbitMQ高级特性：TTL
TTL全称Time To Live (存活时间/过期时间). 当消息到达存活时间后,还没有被消费,会被自动清除.. RabbitMQ可以对消息设置过期时间,也可以对整个队列(Queue)设置过期时间. ...

rabbitmq高级特性（消息手动确认）

为了保证消息不丢失，从生产者两种模式（确认模式和返回模式）到消费者手动确认的一个大整合

生产者

消费者

rabbitmq高级特性（消息手动确认）相关推荐

最新文章

热门文章