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.xdnf.cn/news/143111.html

如若内容造成侵权/违法违规/事实不符，请联系一条长河网进行投诉反馈，一经查实，立即删除！

相关文章

适合企业的跨隔离网文件摆渡的四种方式及优缺点

适合企业的跨隔离网文件摆渡的四种方式及优缺点

在当今的信息化时代&#xff0c;数据是企业的重要资产&#xff0c;如何有效地管理和传输数据&#xff0c;尤其是跨隔离网的大文件数据&#xff0c;是企业面临的一个挑战。跨隔离网文件摆渡是指在不同安全等级或网络环境的计算机之间传输文件的过程。 目前&#xff0c;市场上常见…
阅读更多...
知识图谱01——ubuntu22.04 neo4j安装配置

知识图谱01——ubuntu22.04 neo4j安装配置

环境&#xff1a;ubuntu22.04 以上参照自 目录 下载安装包新建conda虚拟环境启动neo4j错误处理 远程配置文件修改可能出现的错误 远程启动python远程访问(下面内容还未尝试&#xff0c;先写) 下载安装包 进入网站https://neo4j.com/deployment-center/#community 由于是ubunt…
阅读更多...
Java——》IO

Java——》IO

推荐链接&#xff1a; 总结——》【Java】 总结——》【Mysql】 总结——》【Redis】 总结——》【Kafka】 总结——》【Spring】 总结——》【SpringBoot】 总结——》【MyBatis、MyBatis-Plus】 总结——》【Linux】 总结——》【MongoD…
阅读更多...
开利网络受邀参与生态合作伙伴和合控股“数利丰”品牌营销会议

开利网络受邀参与生态合作伙伴和合控股“数利丰”品牌营销会议

近日&#xff0c;开利网络受邀出席生态合作伙伴“数利丰”品牌营销会议&#xff0c;就“数利丰”产品的技术能力和案例沉淀进行分享。 作为“数利丰”项目的技术支持方&#xff0c;开利网络创始人付立军在分享会上表示&#xff0c;现如今&#xff0c;每个企业都至少做过一套系统…
阅读更多...
[论文笔记]Prefix Tuning

[论文笔记]Prefix Tuning

引言 今天带来微调LLM的第二篇论文笔记Prefix-Tuning。 作者提出了用于自然语言生成任务的prefix-tuning(前缀微调)的方法，固定语言模型的参数而优化一些连续的任务相关的向量，称为prefix。受到了语言模型提示词的启发，允许后续的token序列注意到这些prefix，当成虚拟toke…
阅读更多...
【数据结构——单链表】本篇文章通过图文结合的方式能让你轻松的掌握单链表

【数据结构——单链表】本篇文章通过图文结合的方式能让你轻松的掌握单链表

链表的概念及结构 有了顺序表为什么还会出现链表呢&#xff1f; 链表和数组类似&#xff0c;但是功能比数组强大的多&#xff0c;数组的空间是固定的&#xff0c;在定义数组的时候空间大小就已经固定了&#xff0c;在使用时有可能会造成空间的浪费或者面临空间不够的风险&…
阅读更多...
leetcode 23. 合并 K 个升序链表

leetcode 23. 合并 K 个升序链表

2023.9.25 本题要合并k个有序链表&#xff0c;最朴素的方法可以联想到之前做的合并两个有序链表。 用一个for循环遍历lists数组&#xff0c;利用合并两个有序链表的代码&#xff0c;不断合并lists中的链表&#xff0c;最后返回头节点即可。 代码如下&#xff1a; /*** Definit…
阅读更多...
目标分类笔记(二): 利用PaddleClas的框架来完成多标签分类任务（从数据准备到训练测试部署的完整流程）

目标分类笔记(二): 利用PaddleClas的框架来完成多标签分类任务（从数据准备到训练测试部署的完整流程）

文章目录 一、演示多分类效果二、PaddleClas介绍三、代码获取四、数据集获取五、环境搭建六、数据格式分析七、模型训练7.1 模型恢复训练7.2 多卡训练7.3 其他训练指标 八、模型预测九、模型评估十、PaddleClas相关博客 一、演示多分类效果 二、PaddleClas介绍 PaddleClas主要…
阅读更多...
PY32F003F18之RTC

PY32F003F18之RTC

一、RTC振荡器 PY32F003F18实时时钟的振荡器是内部RC振荡器&#xff0c;频率为32.768KHz。它也可以使用HSE时钟&#xff0c;不建议使用。HAL库提到LSE振荡器&#xff0c;但PY32F003F18实际上没有这个振荡器。 缺点&#xff1a;CPU掉电后&#xff0c;需要重新配置RTC&#xff…
阅读更多...
【深度学习】图像去噪（2）——常见网络学习

【深度学习】图像去噪（2）——常见网络学习

【深度学习】图像去噪 是在 【深度学习】计算机视觉 系列文章的基础上&#xff0c;再次针对深度学习&#xff08;尤其是图像去噪方面&#xff09;的基础知识有更深入学习和巩固。 1 DnCNN 1.1 网络结构 1.1.1 残差学习 1.1.2 Batch Normalization (BN) 1.1.2.1 背景和目标…
阅读更多...
java项目之人事管理系统（ssm源码+文档）

java项目之人事管理系统（ssm源码+文档）

项目简介 人事管理系统实现了以下功能&#xff1a; 管理员&#xff1a;个人中心、员工管理、部门经理管理、部门信息管理、员工考勤管理、签到管理、请假申请管理、工资查询管理、部门类型管理.部门经理&#xff1a;个人中心、员工管理、部门信息管理、员工考勤管理、签到管理…
阅读更多...
Baichuan2 技术报告笔记

Baichuan2 技术报告笔记

文章目录 预训练预训练数据模型架构TokenizerPositional EmbeddingsAcitivations and NormalizationsOptimizations 对齐Supervised Fine-TuningRLHF 安全性预训练阶段对齐阶段 参考资料 对Baichuan2技术报告阅读后的笔记 Baichuan2 与其他大模型的对比如下表 预训练 预训练数…
阅读更多...
【Linux】C语言实现对文件的加密算法

【Linux】C语言实现对文件的加密算法

异或加密 解密方式是进行第二次加密后自动解密 #define BUF_SIZE (16384) //16k /************************************************************** 功能描述: 加密实现 输入参数: --------------------------------------------------------------- 修改作者: 修改日期…
阅读更多...
山西电力市场日前价格预测【2023-09-27】

山西电力市场日前价格预测【2023-09-27】

日前价格预测 预测说明&#xff1a; 如上图所示&#xff0c;预测明日&#xff08;2023-09-27&#xff09;山西电力市场全天平均日前电价为342.48元/MWh。其中&#xff0c;最高日前电价为454.24元/MWh&#xff0c;预计出现在18: 30。最低日前电价为171.32元/MWh&#xff0c;预计…
阅读更多...
如何永久关闭WPS任务窗口？

如何永久关闭WPS任务窗口？

1、按住任务窗口上的浮动按钮&#xff0c;将其拖出来成悬浮窗口。 第二步&#xff0c;使用火绒弹窗拦截&#xff0c;选中弹出的窗口&#xff0c;进行拦截。注意&#xff1a;拦截次数为2次。即进行2次操作。 操作两次后&#xff0c;弹窗被拦截&#xff0c;此时Word文档改为双页显…
阅读更多...
蓝桥杯每日一题20223.9.26

蓝桥杯每日一题20223.9.26

4407. 扫雷 - AcWing题库 题目描述 分析 此题目使用map等都会超时&#xff0c;所以我们可以巧妙的使用哈希模拟散列表&#xff0c;哈希表初始化为-1首先将地雷读入哈希表&#xff0c;找到地雷的坐标在哈希表中对应的下标&#xff0c;如果没有则此地雷的位置第一次出现&#…
阅读更多...
QQ怎么上传大于1G的视频啊？视频压缩这样做

QQ怎么上传大于1G的视频啊？视频压缩这样做

当我们想要在QQ上分享一段大容量的视频时&#xff0c;往往会因为超过1G的限制而感到无助。不过&#xff0c;不用担心&#xff0c;今天我们将为你介绍三种可以压缩视频大小的方法&#xff0c;一起来看看吧~ 一、嗨格式压缩大师 嗨格式压缩大师是一款专业的视频压缩软件&#xf…
阅读更多...
全渠道客服体验：Rocket.Chat 的无缝互动 | 开源日报 No.41

全渠道客服体验：Rocket.Chat 的无缝互动 | 开源日报 No.41

RocketChat/Rocket.Chat Stars: 36.9k License: NOASSERTION Rocket.Chat 是一个完全可定制的开源通信平台&#xff0c;适用于具有高标准数据保护要求的组织。我们是团队沟通场景下的最终免费开源解决方案&#xff0c;可以实现同事之间、公司之间或客户之间的实时对话。提高生…
阅读更多...
13. ShardingSphere-Proxy 数据库代理

13. ShardingSphere-Proxy 数据库代理

Spring Cloud 微服务系列文章&#xff0c;点击上方合集↑ 1. 简介 ShardingSphere-Proxy是ShardingSphere分布式数据库中间件的一部分&#xff0c;它提供了数据库代理功能。通过引入ShardingSphere-Proxy&#xff0c;可以在无需改动应用程序代码的情况下&#xff0c;实现分库…
阅读更多...
使用Process Monitor工具探测日志文件是程序哪个模块生成的

使用Process Monitor工具探测日志文件是程序哪个模块生成的

目录 1、问题描述 2、使用Process Monitor监测目标文件是哪个模块生成的思路说明 3、操作Process Monitor监测日志文件是哪个模块生成的 4、通过screenctach.dll库的时间戳&#xff0c;找到其pdb文件&#xff0c;然后去查看详细的函数调用堆栈 5、最后 VC常用功能开发汇总…
阅读更多...
最新文章

Copyright @ 2022~2023