Rabbit消息的可靠性

生产者重连

 消费者重试

 

 

Confirm模式简介

消息的confirm确认机制，是指生产者投递消息后，到达了消息服务器Broker里面的exchange交换机，则会给生产者一个应答，生产者接收到应答，用来确定这条消息是否正常的发送到Broker的exchange中，这也是消息可靠性投递的重要保障；

    1. 具体代码设置

yml 

server:port: 8080spring:application:name: confirm-learn1rabbitmq:host: 192.168.126.130port: 5672username: adminpassword: 123456virtual-host: powernodepublisher-confirm-type: correlated # 开启生产者的确认模式，设置关联模式my:exchangeName: exchange.confirm.1queueName: queue.confirm.1

配置类

package com.powernode.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Value("${my.exchangeName}")private String exchangeName;@Value("${my.queueName}")private String queueName;@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange(exchangeName).build();}@Beanpublic Queue queue(){return QueueBuilder.durable(queueName).build();}@Beanpublic Binding binding(DirectExchange directExchange,Queue queue){return BindingBuilder.bind(queue).to(directExchange).with("info");}}

写法一

 配置回调类

package com.powernode.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;@Component
@Slf4j
public class MyConfirmCallBack implements RabbitTemplate.ConfirmCallback {@Overridepublic void confirm(CorrelationData correlationData, boolean ack, String cause) {log.info("关联id为：{}",correlationData.getId()+"");if (ack){log.info("消息正确的达到交换机");return;}//ack =false 没有到达交换机log.error("消息没有到达交换机，原因为：{}",cause);}
}

 发送消息类

package com.powernode.service;import com.powernode.config.MyConfirmCallBack;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@Resourceprivate MyConfirmCallBack confirmCallBack;@PostConstruct //构造方法后执行它，相当于初始化作用public void init(){rabbitTemplate.setConfirmCallback(confirmCallBack);}public void sendMsg(){Message message= MessageBuilder.withBody("hello world".getBytes()).build();CorrelationData correlationData=new CorrelationData(); //关联数据correlationData.setId("order_123456"); //发送订单信息rabbitTemplate.convertAndSend("exchange.confirm.1","info",message,correlationData);log.info("消息发送完毕，发送时间为：{}",new Date());}
}

 启动类

package com.powernode;import com.powernode.service.MessageService;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;import javax.annotation.Resource;@SpringBootApplication
public class Application implements ApplicationRunner {@Resourceprivate MessageService messageService;public static void main(String[] args) {SpringApplication.run(Application.class, args);}@Overridepublic void run(ApplicationArguments args) throws Exception {messageService.sendMsg();}
}

 方法二

利用lambda 可以省掉配置回调类

package com.powernode.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService{@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstruct //构造方法后执行它，相当于初始化作用public void init(){rabbitTemplate.setConfirmCallback(//lambda 表达式(correlationData, ack, cause)->{log.info("关联id为：{}",correlationData.getId()+"");if (ack){log.info("消息正确的达到交换机");return;}//ack =false 没有到达交换机log.error("消息没有到达交换机，原因为：{}",cause);});}public void sendMsg(){Message message= MessageBuilder.withBody("hello world".getBytes()).build();CorrelationData correlationData=new CorrelationData(); //关联数据correlationData.setId("order_123456"); //发送订单信息rabbitTemplate.convertAndSend("exchange.confirm.4dddd","info",message,correlationData);log.info("消息发送完毕，发送时间为：{}",new Date());}}

  1. RabbitMQ消息Return模式

  1. 消息可靠性投递

rabbitmq 整个消息投递的路径为：

producer —> exchange —> queue —> consumer

>> 消息从 producer 到 exchange 则会返回一个 confirmCallback；

>> 消息从 exchange –> queue 投递失败则会返回一个 returnCallback；

我们可以利用这两个callback控制消息的可靠性投递；

开启 确认模式；

使用rabbitTemplate.setConfirmCallback设置回调函数，当消息发送到exchange后回调confirm方法。在方法中判断ack，如果为true，则发送成功，如果为false，则发送失败，需要处理；

注意配置文件中，开启 退回模式；

spring.rabbitmq.publisher-returns: true

使用rabbitTemplate.setReturnCallback设置退回函数，当消息从exchange路由到

queue失败后，则会将消息退回给producer，并执行回调函数returnedMessage；

yml

server:port: 8080spring:application:name: ttl-learn1rabbitmq:host: 192.168.126.130port: 5672username: adminpassword: 123456virtual-host: powernodepublisher-returns: true #开启return模式my:exchangeName: exchange.return.1queueName: queue.return.1

配置类

package com.powernode.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Value("${my.exchangeName}")private String exchangeName;@Value("${my.queueName}")private String queueName;@Beanpublic DirectExchange directExchange(){return ExchangeBuilder.directExchange(exchangeName).build();}@Beanpublic Queue queue(){return QueueBuilder.durable(queueName).build();}@Beanpublic Binding binding(DirectExchange directExchange,Queue queue){return BindingBuilder.bind(queue).to(directExchange).with("info");}
}

方式一 

发送消息类

package com.powernode.service;import com.powernode.config.MyReturnCallBack;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;@Resourceprivate MyReturnCallBack myReturnCallBack;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(myReturnCallBack); //设置回调}public void sendMsg(){Message message= MessageBuilder.withBody("hello world".getBytes()).build();rabbitTemplate.convertAndSend("exchange.return.1","info1111",message);log.info("消息发送完毕，发送时间为：{}",new Date());}
}

      回调配置类

package com.powernode.config;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;/*** 当消息从交换机 没有正确地 到达队列，则会触发该方法* 如果消息从交换机 正确地 到达队列了，那么就不会触发该方法** @param returned*/
@Component
@Slf4j
public class MyReturnCallBack implements RabbitTemplate.ReturnsCallback {@Overridepublic void returnedMessage(ReturnedMessage returnedMessage) {log.error("消息从交换机没有正确的路由到（投递到）队列，原因为：{}",returnedMessage.getReplyText());}
}

方式二 lambda

package com.powernode.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService{@Resourceprivate RabbitTemplate rabbitTemplate;@PostConstructpublic void init(){rabbitTemplate.setReturnsCallback(//使用lambda表达式message->{log.error("消息从交换机没有正确的路由到（投递到）队列，原因为：{}",message.getReplyText());}); //设置回调}public void sendMsg(){Message message= MessageBuilder.withBody("hello world".getBytes()).build();rabbitTemplate.convertAndSend("exchange.return.4","info",message);log.info("消息发送完毕，发送时间为：{}",new Date());}}

RabbitMQ交换机详细属性

3.1具体参数

1、Name：交换机名称；就是一个字符串

2、Type：交换机类型，direct, topic, fanout, headers四种

3、Durability：持久化，声明交换机是否持久化，代表交换机在服务器重启后是否还存在；

4、Auto delete：是否自动删除，曾经有队列绑定到该交换机，后来解绑了，那就会自动删除该交换机；

5、Internal：内部使用的，如果是yes，客户端无法直接发消息到此交换机，它只能用于交换机与交换机的绑定。

6、Arguments：只有一个取值alternate-exchange，表示备用交换机；

3.2代码演示

结论1：没发消息之前不会创建交换机和对列

结论2：发消息后，如果交换机不存在，才开始创建交换机，如果队列不存在，则创建新的对列

结论3：创建交换机或者队列完成后再重新创建，如果修改交换机或队列参数则会报错

406错误（inequivalent arg 'durable' for exchange 'exchange.durability' in vhost 'powernode': received 'false' but current is 'true', class-id=40, method-id=10)）

结论4：设置持久化为false ，重启rabbitmq-server，则交换机丢失，实验durable参数，先看下控制台，然后重启rabbitmq-server

结论5：实验自动删除为 true ，从控制台上手动解绑，会发现自动删除

3.3 备用交换机

3.3.1 备用交换机使用场景

当消息经过交换器准备路由给队列的时候，发现没有对应的队列可以投递信息，在rabbitmq中会默认丢弃消息，如果我们想要监测哪些消息被投递到没有对应的队列，我们可以用备用交换机来实现，可以接收备用交换机的消息，然后记录日志或发送报警信息。

3.3.2 主要代码和注意事项

备用交换机示例如下：

注意：备用交换机一般使用fanout交换机

测试时：指定一个错误路由

重点：普通交换机设置参数绑定到备用交换机

Map<String, Object> arguments = new HashMap<>();

//指定当前正常的交换机的备用交换机是谁

arguments.put("alternate-exchange", EXCHANGE_ALTERNATE);

//DirectExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments)

return new DirectExchange(EXCHANGE, true, false, arguments);

//return ExchangeBuilder.directExchange(EXCHANGE).withArguments(args).build();

3.3.3 参考配置代码

yml

server:port: 8080spring:application:name: ttl-learn1rabbitmq:host: 192.168.126.130port: 5672username: adminpassword: 123456virtual-host: powernodemy:exchangeNormalName: exchange.normal.alternate  #正常交换机exchangeAlternateName: exchange.alternate.1 #备用交换机queueNormalName: queue.normal.alternate  #正常队列queueAlternateName: queue.alternate.1  # 备用队列

配置类 

package com.powernode.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Value("${my.exchangeNormalName}")private String exchangeNormalName;@Value("${my.exchangeAlternateName}")private String exchangeAlternateName;@Value("${my.queueNormalName}")private String queueNormalName;@Value("${my.queueAlternateName}")private String queueAlternateName;@Beanpublic DirectExchange normalExchange(){return ExchangeBuilder // 默认为持久化的，默认不自动删除.directExchange(exchangeNormalName) // 交换机的名字.alternate(exchangeAlternateName) //设置备用交换机 alternate-exchange.build();}@Beanpublic Queue queueNormal(){return QueueBuilder.durable(queueNormalName).build();}@Beanpublic Binding binding(DirectExchange normalExchange,Queue queueNormal){return BindingBuilder.bind(queueNormal).to(normalExchange).with("info");}@Bean //备用交换机public FanoutExchange alternateExchange(){return ExchangeBuilder.fanoutExchange(exchangeAlternateName).build();}@Beanpublic Queue alternateQueue(){return QueueBuilder.durable(queueAlternateName).build();}@Beanpublic Binding bindingAlternate(FanoutExchange alternateExchange,Queue alternateQueue){return BindingBuilder.bind(alternateQueue).to(alternateExchange);}
}

3.3.4 参考发送消息代码

@Service

public class MessageService {

    @Resource

    private RabbitTemplate rabbitTemplate;

    /**

     * 发送消息

     */

public void sendMessage() {

//我们故意写错路由key,由于我们正常交换机设置了备用交换机，所以该消息就会进入备用交换机
//从而进入备用对列，我们可以写一个程序接收备用对列的消息，接收到后通知相关人员进行处理
//如果正常交换机没有设置备用交换机，则该消息会被抛弃。

        rabbitTemplate.convertAndSend(RabbitConfig.EXCHANGE, "info1223", "hello");

        System.out.println("消息发送完毕......");

    }

}

RabbitMQ队列详细属性

4.1 具体参数

Type：队列类型

Name：队列名称，就是一个字符串，随便一个字符串就可以；

Durability：声明队列是否持久化，代表队列在服务器重启后是否还存在；

Auto delete： 是否自动删除，如果为true，当没有消费者连接到这个队列的时候，队列会自动删除；

Exclusive：exclusive属性的队列只对首次声明它的连接可见，并且在连接断开时自动删除；

基本上不设置它，设置成false

Arguments：队列的其他属性，例如指定DLX（死信交换机等）；

1、x-expires：Number

当Queue（队列）在指定的时间未被访问，则队列将被自动删除；

2、x-message-ttl：Number

发布的消息在队列中存在多长时间后被取消（单位毫秒）；

3、x-overflow：String

设置队列溢出行为，当达到队列的最大长度时，消息会发生什么，有效值为Drop Head或Reject Publish；

4、x-max-length：Number

队列所能容下消息的最大长度，当超出长度后，新消息将会覆盖最前面的消息，类似于Redis的LRU算法；

5、 x-single-active-consumer：默认为false

激活单一的消费者，也就是该队列只能有一个消息者消费消息；

6、x-max-length-bytes：Number

限定队列的最大占用空间，当超出后也使用类似于Redis的LRU算法；

7、x-dead-letter-exchange：String

指定队列关联的死信交换机，有时候我们希望当队列的消息达到上限后溢出的消息不会被删除掉，而是走到另一个队列中保存起来；

8.x-dead-letter-routing-key：String

指定死信交换机的路由键，一般和6一起定义；

9.x-max-priority：Number

如果将一个队列加上优先级参数，那么该队列为优先级队列；

（1）、给队列加上优先级参数使其成为优先级队列

x-max-priority=10【0-255取值范围】

（2）、给消息加上优先级属性

通过优先级特性，将一个队列实现插队消费；

MessageProperties messageProperties=new MessageProperties();
messageProperties.setPriority(8);

10、x-queue-mode：String（理解下即可）

队列类型x-queue-mode=lazy懒队列，在磁盘上尽可能多地保留消息以减少RAM使用，如果未设置，则队列将保留内存缓存以尽可能快地传递消息；

11、x-queue-master-locator：String（用的较少，不讲）

在集群模式下设置队列分配到的主节点位置信息；

每个queue都有一个master节点，所有对于queue的操作都是事先在master上完成，之后再slave上进行相同的操作；

每个不同的queue可以坐落在不同的集群节点上，这些queue如果配置了镜像队列，那么会有1个master和多个slave。

基本上所有的操作都落在master上，那么如果这些queues的master都落在个别的服务节点上，而其他的节点又很空闲，这样就无法做到负载均衡，那么势必会影响性能；

关于master queue host 的分配有几种策略，可以在queue声明的时候使用x-queue-master-locator参数，或者在policy上设置queue-master-locator，或者直接在rabbitmq的配置文件中定义queue_master_locator，有三种可供选择的策略：

（1）min-masters：选择master queue数最少的那个服务节点host；

（2）client-local：选择与client相连接的那个服务节点host；

（3）random：随机分配；

4.2 参考代码

@Configuration

public class RabbitConfig {

    public static final String EXCHANGE = "exchange";

    public static final String QUEUE = "queue";

    public static final String KEY = "info";

    QueueBuilder builder;

    @Bean

    public DirectExchange directExchange() {

        return ExchangeBuilder.directExchange(EXCHANGE).build();

    }

    @Bean

    public Queue queue() {

        Map<String, Object> arguments = new HashMap<>();

        //arguments.put("x-expires", 5000);

        //arguments.put("x-max-length", 5);

        //arguments.put("x-overflow", "reject-publish");

        arguments.put("x-single-active-consumer", false); //TODO ???

        //arguments.put("x-max-length-bytes", 20); // 单位是字节

        //arguments.put("x-max-priority", 10); // 0-255 //表示把当前声明的这个队列设置成了优先级队列，那么该队列它允许消息插队

        //将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM内存的使用，如果未设置，队列将保留内存缓存以尽可能快地传递消息；

        //有时候我们把这种队列叫：惰性队列

        //arguments.put("x-queue-mode", "lazy");

        //设置队列版本。默认为版本1。

        //版本1有一个基于日志的索引，它嵌入了小消息。

        //版本2有一个不同的索引，可以在许多场景中提高内存使用率和性能，并为以前嵌入的消息提供了按队列存储。

        //arguments.put("x-queue-version", 2);

        // x-queue-master-locator：在集群模式下设置镜像队列的主节点信息。

        //arguments.put("x-queue-master-locator", QueueBuilder.LeaderLocator.clientLocal.getValue());

        //-------------------------

        //arguments.put("x-expires", 10000); //自动过期，10秒

        //arguments.put("x-message-ttl", 10000); //自动过期，10秒，不会删除队列

        //QueueBuilder 类里面有定义，设置队列溢出行为，当达到队列的最大长度时消息会发生什么，有效值是drop-head、reject-publish

        //arguments.put("x-max-length", 5);

        //arguments.put("x-overflow", QueueBuilder.Overflow.dropHead.getValue());

        //表示队列是否是单一活动消费者，true时，注册的消费组内只有一个消费者消费消息，其他被忽略，false时消息循环分发给所有消费者(默认false)

        //arguments.put("x-single-active-consumer", true);

        // x-max-length-bytes，队列消息内容占用最大空间，受限于内存大小，超过该阈值则从队列头部开始删除消息；

        //arguments.put("x-max-length-bytes", 10);

        //参数是1到255之间的正整数，表示队列应该支持的最大优先级，数字越大代表优先级越高，没有设置priority优先级字段，那么priority字段值默认为0；如果优先级队列priority属性被设置为比x-max-priority大，那么priority的值被设置为x-max-priority的值。

        //arguments.put("x-max-priority", 10);

        //将队列设置为延迟模式，在磁盘上保留尽可能多的消息，以减少RAM的使用;如果未设置，队列将保留内存缓存以尽可能快地传递消息；

        //arguments.put("x-queue-mode", "lazy");

        arguments.put("x-queue-version", 2);

        // x-queue-master-locator：在集群模式下设置镜像队列的主节点信息。

        arguments.put("x-queue-master-locator", QueueBuilder.LeaderLocator.clientLocal.getValue());

        //---------------------------------------------

        // Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, @Nullable Map<String, Object> arguments)

        return new Queue(QUEUE, true, false, false, arguments);

    }

    @Bean

    public Binding binding(DirectExchange directExchange, Queue queue) {

        return BindingBuilder.bind(queue).to(directExchange).with(KEY);

    }

}

实验durable 参数 重启rabbitmq-server，队列丢失

实验autodelete参数：加入接收者，发现停掉服务，那么久没有消费者了，对列就会自动删除

消息可靠性投递

消息的可靠性投递就是要保证消息投递过程中每一个环节都要成功，那么这肯定会牺牲一些性能，性能与可靠性是无法兼得的；

如果业务实时一致性要求不是特别高的场景，可以牺牲一些可靠性来换取性能。

确保消息在队列正确地存储

可能因为系统宕机、重启、关闭等等情况导致存储在队列的消息丢失，即③出现问题；

解决方案：

  1. 、队列持久化

代码：

QueueBuilder.durable(QUEUE).build();

  1. 、交换机持久化

代码：

ExchangeBuilder.directExchange(EXCHANGE).durable(true).build();

  1. 、消息持久化

代码：

默认持久化

 MessageProperties messageProperties = new MessageProperties();

//设置消息持久化，当然它默认就是持久化，所以可以不用设置，可以查看源码

 messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);

  1. 、集群，镜像队列，高可用

  1. 确保消息从队列正确地投递到消费者

采用消息消费时的手动ack确认机制来保证；

如果消费者收到消息后未来得及处理即发生异常，或者处理过程中发生异常，会导致④失败。

为了保证消息从队列可靠地达到消费者，RabbitMQ提供了消息确认机制（message acknowledgement）；

开启手动ack消息消费确认


spring.rabbitmq.listener.simple.acknowledge-mode=manual

 yml

server:port: 8080spring:application:name: rabbit-12-reliabilityrabbitmq:host: 192.168.126.130port: 5672username: adminpassword: 123456virtual-host: powernodepublisher-confirm-type: correlated # 开启发布者的确认模式publisher-returns: true # 开启发布者的return模式listener:simple:acknowledge-mode: manual # 开始消费者的手动确认模式my:exchangeName: exchange.reliabilityqueueName: queue.reliability

 配置类

package com.powernode.config;import org.springframework.amqp.core.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RabbitConfig {@Value("${my.exchangeName}")private String exchangeName;@Value("${my.queueName}")private String queueName;@Beanpublic DirectExchange directExchange(){//默认就是持久化的return ExchangeBuilder.directExchange(exchangeName).build();}@Beanpublic Queue queue(){//队列持久化return QueueBuilder.durable(queueName).build();}@Beanpublic Binding binding(DirectExchange directExchange,Queue queue){return BindingBuilder.bind(queue).to(directExchange).with("info");}
}

消息类

package com.powernode.service;import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageBuilder;
import org.springframework.amqp.core.MessageDeliveryMode;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Service;import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.Date;@Service
@Slf4j
public class MessageService {@Resourceprivate RabbitTemplate rabbitTemplate;/*** 构造方法执行后自动执行*/@PostConstructpublic void init(){//开启生产者的确定模式rabbitTemplate.setConfirmCallback((correlationData, ack, cause)->{if(!ack){log.error("消息没有到达交换机，原因为：{}",cause);//TODO 重发消息或者记录错误日志}});rabbitTemplate.setReturnsCallback(returnedMessage->{log.error("消息没有从交换机正确的投递（路由）到队列，原因为：{}",returnedMessage.getReplyText());//TODO 记录错误日志，给程序员发短信或者或者邮件});}public void sendMsg(){MessageProperties messageProperties=new MessageProperties();//设置单条消息的持久化，默认就是持久化messageProperties.setDeliveryMode(MessageDeliveryMode.PERSISTENT);Message message= MessageBuilder.withBody("hello world".getBytes()).andProperties(messageProperties).build();rabbitTemplate.convertAndSend("exchange.reliability","info",message);log.info("消息发送完毕，发送时间为：{}",new Date());}
}

发送消息

package com.powernode;import com.powernode.service.MessageService;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;import javax.annotation.Resource;@SpringBootApplication
public class Application implements ApplicationRunner {@Resourceprivate MessageService messageService;public static void main(String[] args) {SpringApplication.run(Application.class, args);}@Overridepublic void run(ApplicationArguments args) throws Exception {messageService.sendMsg();}
}

 消费者消费消息类（手动确认）

package com.powernode.message;import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;import java.io.IOException;@Component
@Slf4j
public class ReceiveMessage {@RabbitListener(queues = {"queue.reliability"})public void receiveMsg(Message message, Channel channel){//获取消息的唯一标识long deliveryTag = message.getMessageProperties().getDeliveryTag();try {log.info("接收到的消息为：{}",new String(message.getBody()));// TODO 插入订单等
//            int a=10/0;//手动确认channel.basicAck(deliveryTag,false);} catch (Exception e) {log.error("消息处理出现问题");try {channel.basicNack(deliveryTag,false,true);} catch (IOException ex) {throw new RuntimeException(ex);}throw new RuntimeException(e);}}
}

消费者在订阅队列时，通过上面的配置，不自动确认，采用手动确认，RabbitMQ会等待消费者显式地回复确认信号后才从队列中删除消息；

如果消息消费失败，也可以调用basicReject()或者basicNack()来拒绝当前消息而不是确认。如果requeue参数设置为true，可以把这条消息重新存入队列，以便发给下一个消费者（当然，只有一个消费者的时候，这种方式可能会出现无限循环重复消费的情况，可以投递到新的队列中，或者只打印异常日志）；

  1. 消息的幂等性

消息消费时的幂等性（消息不被重复消费）

同一个消息，第一次接收，正常处理业务，如果该消息第二次再接收，那就不能再处理业务，否则就处理重复了；

幂等性是：对于一个资源，不管你请求一次还是请求多次，对该资源本身造成的影响应该是相同的，不能因为重复的请求而对该资源重复造成影响；

以接口幂等性举例：

接口幂等性是指：一个接口用同样的参数反复调用，不会造成业务错误，那么这个接口就是具有幂等性的；

注册接口；

发送短信验证码接口；

比如同一个订单我支付两次，但是只会扣款一次，第二次支付不会扣款，这说明这个支付接口是具有幂等性的；

如何避免消息的重复消费问题？（消息消费时的幂等性）

全局唯一ID + Redis

生产者在发送消息时，为每条消息设置一个全局唯一的messageId，消费者拿到消息后，使用setnx命令，将messageId作为key放到redis中：setnx(messageId, 1)，若返回1，说明之前没有消费过，正常消费；若返回0，说明这条消息之前已消费过，抛弃；

具体代码参考以下代码；

参考代码：

  //1、把消息的唯一ID写入redis

        boolean flag = stringRedisTemplate.opsForValue().setIfAbsent("idempotent:" + orders.getId(), String.valueOf(orders.getId())); //如果redis中该key不存在，那么就设置，存在就不设置

        if (flag) { //key不存在返回true

            //相当于是第一次消费该消息

            //TODO 处理业务

            System.out.println("正常处理业务....." + orders.getId());

        }

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/88763.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

【大数据】Doris 构建实时数仓落地方案详解（三）：Doris 实时数仓设计

【大数据】Doris 构建实时数仓落地方案详解（三）：Doris 实时数仓设计

本系列包含&#xff1a; Doris 构建实时数仓落地方案详解&#xff08;一&#xff09;&#xff1a;实时数据仓库概述Doris 构建实时数仓落地方案详解&#xff08;二&#xff09;&#xff1a;Doris 核心功能解读Doris 构建实时数仓落地方案详解&#xff08;三&#xff09;&#…
阅读更多...
数据库存储引擎和数据类型详细介绍

数据库存储引擎和数据类型详细介绍

目录 一、数据库存储引擎&#xff08;了解&#xff09;1.了解MySQL体系结构2.存储引擎&#xff08;了解&#xff09;2.1.存储引擎的介绍2.2.存储引擎分类2.3.如何选择引擎&#xff1f; 3.事务控制语言(TCL)事务的四个特性(ACID) 二、数据类型&#xff08;了解&#xff09;1.整型…
阅读更多...
Servlet操作与用法（保姆式教学）

Servlet操作与用法（保姆式教学）

Servlet介绍 什么是servlet Servlet&#xff08;Servlet Applet的缩写&#xff0c;全称 Java Servlet&#xff09;&#xff1a;适用于Java编写的服务器程序&#xff0c;其主要功能是在于交互式的浏览和修改数据&#xff0c;生成动态Web内容。狭义的Servlet是指Java语言实现的…
阅读更多...
Xmake v2.8.3 发布，改进 Wasm 并支持 Xmake 源码调试

Xmake v2.8.3 发布，改进 Wasm 并支持 Xmake 源码调试

Xmake 是一个基于 Lua 的轻量级跨平台构建工具。 它非常的轻量&#xff0c;没有任何依赖&#xff0c;因为它内置了 Lua 运行时。 它使用 xmake.lua 维护项目构建&#xff0c;相比 makefile/CMakeLists.txt&#xff0c;配置语法更加简洁直观&#xff0c;对新手非常友好&#x…
阅读更多...
ARM day1

ARM day1

1.复习今日内容 2.搭建汇编环境 下发资料-》工具软件 -》汇编环境搭建 3.安装Ubuntu下的交叉编译工具链 思维导图&#xff1a;
阅读更多...
latexocr安装过程中遇到的问题解决办法

latexocr安装过程中遇到的问题解决办法

环境要求&#xff1a;需要Python版本3.7&#xff0c;并安装相应依赖文件 具体的详细安装步骤可见我上次写的博文&#xff1a;Mathpix替代者|科研人必备公式识别插件|latexocr安装教程 ‘latexocr‘ 不是内部或外部命令&#xff0c;也不是可运行的程序或批处理文件的相关解决办…
阅读更多...
十五、异常（2）

十五、异常（2）

本章概要 自定义异常 异常与记录日志 异常声明 自定义异常 不必拘泥于 Java 已有的异常类型。Java异常体系不可能预见你将报告的所有错误&#xff0c;所以你可以创建自己的异常类&#xff0c;来表示你的程序中可能遇到的问题。 要自己定义异常类&#xff0c;必须从已有的异…
阅读更多...
什么是 Redis?

什么是 Redis?

Redis 是一种基于内存的数据库&#xff0c;对数据的读写操作都是在内存中完成的&#xff0c;因此读写速度非常快&#xff0c;常用于缓存&#xff0c;消息队列&#xff0c;分布式锁等场景。 Redis 提供了多种数据类型来支持不同的业务场景&#xff0c;比如 String(字符串)、Has…
阅读更多...
C语言环境搭建(Win)

C语言环境搭建(Win)

一、C语言简介 1、 C语言简介 C语言是一门通用的、面向过程式的编译型语言&#xff0c;它的运行速度极快&#xff0c;仅次于汇编语言。 C语言是计算机产业的核心程序设计语言&#xff0c;操作系统、硬件驱动、关键组件、数据库等都离不开C语言&#xff0c;广泛应用于底层开发。…
阅读更多...
CompletableFuture-链式语法和join方法介绍

CompletableFuture-链式语法和join方法介绍

2.4 案例精讲-从电商网站的比价需求展开 2.4.1 函数式编程已成为主流 Lambda表达式Stream流式调用Chain链式调用Java8函数式编程 函数式接口&#xff1a; 小结&#xff1a; 函数式接口&#xff1a; Java8新特性_四大内置核心函数式接口_java8 内置核心接口_ZHOU_VIP的博客-…
阅读更多...
Nginx之memcached_module模块解读

Nginx之memcached_module模块解读

目录 基本介绍 安装添加模块 模块配置指令 基本介绍 nginx的memcached_module模块可以直接从memcached服务器中读取内容后输出&#xff0c;后续的请求不再经过应用程序处理&#xff0c;如php-fpm、django&#xff0c;大大的提升动态页面的速度。nginx只负责从memcach…
阅读更多...
学习路之PHP--lumen安装配置

学习路之PHP--lumen安装配置

一、下载lumen源码 composer create-project --prefer-dist laravel/lumen blog 安装lumen-generator composer require flipbox/lumen-generator 二、配置 bootstrap\app.php 97行 $app->register(Flipbox\LumenGenerator\LumenGeneratorServiceProvider::class);三、生成…
阅读更多...
【IDEA】IDEA 单行注释开头添加空格

【IDEA】IDEA 单行注释开头添加空格

操作 打开 IDEA 的 Settings 对话框&#xff08;快捷键为CtrlAltS&#xff09;&#xff1b;在左侧面板中选择Editor -> Code Style -> Java&#xff1b;在右侧面板中选择Code Generation选项卡&#xff1b;将Line comment at first column选项设置为false使注释加在行开…
阅读更多...
ICCV 2023|Occ2Net，一种基于3D 占据估计的有效且稳健的带有遮挡区域的图像匹配方法...

ICCV 2023|Occ2Net，一种基于3D 占据估计的有效且稳健的带有遮挡区域的图像匹配方法...

本文为大家介绍一篇入选ICCV 2023的论文&#xff0c;《Occ2Net: Robust Image Matching Based on 3D Occupancy Estimation for Occluded Regions》&#xff0c; 一种基于3D 占据估计的有效且稳健的带有遮挡区域的图像匹配方法。 论文链接&#xff1a;https://arxiv.org/abs/23…
阅读更多...
学生宿舍管理系统（前端java+后端Vue）实现-含前端与后端程序

学生宿舍管理系统（前端java+后端Vue）实现-含前端与后端程序

界面介绍 登录 ###宿舍管理 ###菜单管理 ###角色管理 ###班级管理
阅读更多...
TensorFlow入门(五、指定GPU运算)

TensorFlow入门(五、指定GPU运算)

一般情况下,下载的TensorFlow版本如果是GPU版本,在运行过程中TensorFlow能自动检测。如果检测到GPU,TensorFlow会默认利用找到的第一个GPU来执行操作。如果机器上有超过一个可用的GPU,除第一个之外的其他GPU默认是不参与计算的。如果想让TensorFlow使用这些GPU执行操作,需要将运…
阅读更多...
源码：TMS FlexCel Studio for .NET 7.19

源码：TMS FlexCel Studio for .NET 7.19

TMS FlexCel Studio for .NET 是100% 托管代码 Excel 文件操作引擎以及 Excel 和 PDF 报告生成&#xff0c;适用于 .NET、Xamarin.iOS、Xamarin.Android、Xamarin.Mac、Windows Phone 和 Windows Store 功能概述 使用 FlexCel Studio for .NET 创建可动态快速读写 Excel 文件的…
阅读更多...
多线程(虚拟地址空间)

多线程(虚拟地址空间)

代码展示线程 既然我们提到了&#xff0c;线程隶属于进程&#xff0c;是进程的一个执行分支 真的是这样吗&#xff1f; 我们还需要用代码来验证 初步思路是创建三个线程&#xff0c;其中main函数里面的为主线程 不断循环&#xff0c;并且打印相应的pid 假如它们属于不同的进程…
阅读更多...
Java 大厂八股文面试专题-JVM相关面试题 类加载器

Java 大厂八股文面试专题-JVM相关面试题 类加载器

Java 大厂八股文面试专题-设计模式 工厂方法模式、策略模式、责任链模式-CSDN博客 JVM相关面试题 1 JVM组成 1.1 JVM由那些部分组成&#xff0c;运行流程是什么&#xff1f; 难易程度&#xff1a;☆☆☆ 出现频率&#xff1a;☆☆☆☆ JVM是什么 Java Virtual Machine Java程序…
阅读更多...
JDK、JRE 和 JVM 的区别和联系

JDK、JRE 和 JVM 的区别和联系

三者关系 就这三者的关系而言&#xff0c;jvm是jre的子集&#xff0c;jre是jdk的子集&#xff0c;具体关系如下图&#xff1a; Java的执行流程 对于一个Java程序&#xff0c;其执行流程大致如下&#xff1a; 开发人员使用JDK编写和编译Java源代码&#xff0c;生成Java字节码文…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023