如何限制消费者消费

消费时指定channel.basicQos(0,1,false);

    @RabbitListener(queues = "boot_queue")
    public void getMessage(Message message, Channel channel) throws IOException {
        long qps = 10;
        // prefetchSize(消息大小,0是不限制)、prefechSize(消息数量)、global(true 表示此通道的消费者都适用此策略,false代表仅当前消费者)
        channel.basicQos(0,qps,false);
        System.out.println(message); // 这里 不只是输出 单个 发送的信息,而是 全部输出 消息里面的内容数据
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
    }

如何保证消息投递

消息投递到过程:生产者 -> 交换机 -> 队列

因为上述消息传递过程是2个环节,rabbitmq对上述2个环节提供了2个消息投递确认对机制

  1. confirm 确认模式:消息发送到交换器Exchange后触发回调。每次投递都触发
  2. return 退回模式 :当消息未投递到queue时的反馈。投递成功回调不触发

如果开启上述2种模式报错,可能是 java.lang.IllegalStateException: Only one ConfirmCallback is supported by each RabbitTemplate 解决方案如下:

    // 自己添加多例Bean
    @Bean
    @Scope("prototype")
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        template.setMandatory(true);
        template.setMessageConverter(new SerializerMessageConverter());
        return template;
    }


// 请在Controlle或Service类上添加注解 @Scope(prototype)
@Scope(prototype)

confirm 确认模式

消息发送时,每次都会有一个confirm回调函数,我们在confirm回调函数处理即可

# 开启发送端confirm,高AMQP版本就默认开启,且不允许添加此配置
spring.rabbitmq.publisher-confirms=true
    @RequestMapping("/testSendConfirm")
    public void confirm(){
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String s) {
                if (ack){
                    log.info("发送成功");
                }else {
                    log.error("发送失败!需要执行重新发送消息");
                }
            }
        });
        // 发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.hello","这是整合:testSend方法发送的消息");
    }

return退回模式,当消息未投递到queue时的反馈。

# 发送者开启return模式
spring.rabbitmq.publisher-returns=true
    @RequestMapping("/testSendReturn")
    public void returnTest(){
        // 开启交换机处理失败消息 交换机无法将消息进行路由时,会将该消息返回给生产者
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback(){
            @Override
            public void returnedMessage(Message message, int errorCode, String errorInfo, String exchange, String routeKey) {
                log.info("Return 正在执行");
            }
        });
        // 发送消息
        rabbitTemplate.convertAndSend(RabbitMQConfig.EXCHANGE_NAME,"boot.hello","这是整合:testSend方法发送的消息");
    }

测试confirm 与 return模式,

confirm是消息到交换机的无论成功与失败都触发,如果需要测试发送失败,直接指定假的交换机即可

return是交换机到队列,我们只要指定假的路由,这样交换机到消息就无法发送到队列,就会触发rerurn模式。如果消息发送成功,return回调则不会触发。

如何保证消息消费

手动开启ACK模式

配置文件添加

# 消费者开启手动确认。默认是自动确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual
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;

@Slf4j
@Component
public class RabbitMQListen01 {

    @RabbitListener(queues = "boot_queue")
    public void getMessage(Message message, Channel channel) throws IOException {
        int qps = 1;
        try {
            channel.basicQos(0, qps, false);
            log.info("拿到消息的内容:{}", message); // 这里 不只是输出 单个 发送的信息,而是 全部输出 消息里面的内容数据
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
            // 确认ACK执行后,必须当getMessage方法结束后,才能读取下一条
        } catch (Exception e) {
            // 如果有异常,主动拒绝此消息
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
            log.error("Error..");
        }
    }
}

每当消费者消费完成是,确定一下即可!

消息幂等性

这个问题还是建议兄弟从发送端的confirm、return机制入手,然后消费端的ACK确认处理,如果消费者遇到问题,可以考虑到死信队列,然后针对死信队列进行监听。

面试管也可能会从重复发送、重复消费处理。重复发送,一般来说无所谓。因为我们真正操作我们的业务是消费者。我们发送消息的时候,指定消息id即可。相同消息,我们记录是否消费过,间接解决消息重复消费问题。

乐观锁:通过DB + version字段实现

比如我不小心发送了2条消息,内容一样,扣除用户Money。我不可能扣2次。

update user_wallect set money = money -1,version = version + 1 where user = ‘zhangsan’ and version = 1

第二次消费的时候就不会有version = 1的了,也就实现无重复消费,

Redis也可以实现,Redis的exist判断消息的id,就间接实现无重复消费

本站所有资源版权均属于原作者所有,这里所提供资源均只能用于参考学习用,请勿直接商用。若由于商用引起版权纠纷,一切责任均由使用者承担。更多说明请参考 VIP介绍。

最常见的情况是下载不完整: 可对比下载完压缩包的与网盘上的容量,若小于网盘提示的容量则是这个原因。这是浏览器下载的bug,建议用百度网盘软件或迅雷下载。 若排除这种情况,可在对应资源底部留言,或联络我们。

对于会员专享、整站源码、程序插件、网站模板、网页模版等类型的素材,文章内用于介绍的图片通常并不包含在对应可供下载素材包内。这些相关商业图片需另外购买,且本站不负责(也没有办法)找到出处。 同样地一些字体文件也是这种情况,但部分素材会在素材包内有一份字体下载链接清单。

如果您已经成功付款但是网站没有弹出成功提示,请联系站长提供付款信息为您处理

源码素材属于虚拟商品,具有可复制性,可传播性,一旦授予,不接受任何形式的退款、换货要求。请您在购买获取之前确认好 是您所需要的资源