投稿
  1. 首页
  2. 随笔

RabbitMQ生产方式和解决消息可靠性投递及其他问题

羚羊飞渡 • 2022-12-16 • 随笔 • 阅读 22

RabbitMQ生产方式和解决消息可靠性投递及其他问题,第1张

RabbitMQ生产方式和解决消息可靠性投递及其他问题

RabbitMQ生产方式和解决消息可靠性投递及其他问题
  • 3种Exchange(交换机)与对应生产方式
  • Queue队列重载参数
  • RabbitMQ生产方式
    • Work queues(工作模式)
    • Routing(路由模式)
    • Publish/Subscribe(订阅模式)
    • Topics(主题模式)
  • 消息投递流程
  • 考虑以下问题
    • 如何保证消息没有重复消费
    • 消息可靠性投递
      • 生产端
        • /confirm/i确认模式(保证消息从生产者投递到交换机)
        • return退回模式(保证消息从交换机投递到队列)
      • 消费端
    • 怎么保证消息传递的顺序
    • 一致性问题

3种Exchange(交换机)与对应生产方式
  1. direct
    将消息发送到与交换机绑定且对应routingkey的队列
    生产方式有：Work queues(工作模式)、Routing(路由模式)
  2. fanout
    将消息发送到与交换机绑定的队列，不需要routingkey
    生产方式有：Publish/Subscribe(订阅模式)
  3. topic
    将消息发送到与交换机绑定的队列，routingkey模糊匹配
    生产方式有：Topics(主题模式)
Queue队列重载参数 
durable：是否持久化，true 交换机或者队列会存到本地文件数据库，当mq重启，依然在，false的话，重启或者断电，就没了；默认true
autoDelete：是否自动删除，true当没有Consumer消费者时候，自动删除掉；默认false
exclusive:是否独占：true的话只有一个消费者监听这个队列。默认false

    
    @Bean
    public Queue directQueue() {
        return new Queue(DIRECT_QUEUE);
    }

    
    @Bean
    public Queue directQueue2() {
        return new Queue(DIRECT_QUEUE, true, false, false);
    }
RabbitMQ生产方式
  • Work queues(工作模式) 
    一个或者多个消费者共同消费一个队列中的消息(队列中的每一个消息只可能被其中一个消费者消费)

  • Routing(路由模式) 
    跟订阅模式类似，只不过在订阅模式的基础上加上了类型，订阅模式是分发到所有绑定到交换机的队列，路由模式

  • Publish/Subscribe(订阅模式) 
    生产者生产的消息，所有订阅过的消费者都能够接收到消息(队列中的每一个消息可以被多个消费者消费)

  • Topics(主题模式) 
    主题模式和路由模式很像，路由模式是精确匹配，主题模式是模糊匹配，* 匹配一个单词，# 匹配零个或者多个单词

消息投递流程 
Producer(生产者)=>Channel(网络信道)=>Exchange(交换机)=>Queue(队列，到此会进行存储，即完成投递)=>Channel=>Consumer(消费者)
考虑以下问题

本文着重讲解消息可靠性投递

  1. 如何保证消息没有重复消费 
    保证消费者的幂等性
  2. 消息可靠性投递 生产端 
    /confirm/i确认模式和return退回模式(一般是一起使用)
    /confirm/i确认模式(保证消息从生产者投递到交换机) 
    生产者消息投递后，如果Broker收到消息，则会给生产者一个应答(ack)，生产者进行接收应答，用来确定这条消息是否正常的发送到Broker

    实现如下：
    1.配置文件：spring.rabbitmq.publisher-/confirm/i-type=correlated

    server:
  port: 80
  
spring:
  rabbitmq:
    host: 192.168.31.6
    port: 5672
    username: linhy
    password: 123456
    virtual-host: /
    # confirm 确认模式
    publisher-/confirm/i-type: correlated

    2.消息发送类实现RabbitTemplate./confirm/iCallback接口，实现/confirm/i方法
    3.初始化RabbitTemplate，rabbitTemplate.setConfirmCallback(this);
    4.发送消息new一个CorrelationData对象将主键ID放入，将该对象一起发送(/confirm/i方法参数是一个消息唯一标识，即CorrelationData对象，所以需要将主键ID放入)
    5.判断/confirm/i方法ack参数进行补发 *** 作
    消息发送类具体代码示例

    public class RabbitMqServiceImpl implements RabbitMqService, RabbitTemplate./confirm/iCallback {

    @Resource
    private AmqpTemplate amqpTemplate;

    @Resource
    private RabbitTemplate rabbitTemplate;

    
    @PostConstruct
    public void init() {
        rabbitTemplate.set/confirm/iCallback(this);
    }

    
    @Override
    public void sendMessage(String message) {
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId("12050");
        rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ACK_DIRECT_ROUTING_KEY, message, correlationData);// 消息推送失败return
    }

    
    @Override
    public void /confirm/i(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("消息已被推送=========>" + correlationData);
        if (ack) {
            System.out.println("交换机接受消息成功" + cause);
        } else {
            System.out.println("交换机接受消息失败" + cause);
            // 消息补发 *** 作
            System.out.println("失败id:" + correlationData.getId());
        }
    }

}
    return退回模式(保证消息从交换机投递到队列) 
    用于处理交换机不存在或指定的路由key路由不到的情况

    实现如下：
    1.配置文件：publisher-returns: true(默认为false，即消息路由不可达时消息丢弃；true消息回退)

    server:
  port: 80
  
spring:
  rabbitmq:
    host: 192.168.31.6
    port: 5672
    username: linhy
    password: 123456
    virtual-host: /
    # confirm 确认模式
    publisher-/confirm/i-type: correlated
    # return 退回模式
    publisher-returns: true

    2.消息发送类实现RabbitTemplate.ReturnCallback接口，实现returnedMessage方法
    3.初始化RabbitTemplate，rabbitTemplate.setReturnCallback(this);
    4.获取主键ID进行日志纪录或其他 *** 作(returnedMessage方法参数是消息主体，生产者发送时可将主键ID作为消息主体发送)
    消息发送类具体代码示例

    public class RabbitMqServiceImpl implements RabbitMqService, RabbitTemplate./confirm/iCallback, RabbitTemplate.ReturnCallback {

    @Resource
    private AmqpTemplate amqpTemplate;

    @Resource
    private RabbitTemplate rabbitTemplate;

    
    @PostConstruct
    public void init() {
        rabbitTemplate.set/confirm/iCallback(this);
        rabbitTemplate.setReturnCallback(this);
    }

    
    @Override
    public void sendMessage(String message) {
        CorrelationData correlationData = new CorrelationData();
        correlationData.setId("12050");
        rabbitTemplate.convertAndSend(RabbitMQConfig.DIRECT_EXCHANGE, RabbitMQConfig.ACK_DIRECT_ROUTING_KEY, message, correlationData);// 消息推送失败return
    }

    
    @Override
    public void /confirm/i(CorrelationData correlationData, boolean ack, String cause) {
        System.out.println("消息已被推送=========>" + correlationData);
        if (ack) {
            System.out.println("交换机接受消息成功" + cause);
        } else {
            System.out.println("交换机接受消息失败" + cause);
            // 消息补发 *** 作
            System.out.println("失败id:" + correlationData.getId());
        }
    }

    
    @Override
    public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
        System.out.println("消息推送失败，return方法被执行...");
        System.out.println("消息主体：" + new String(message.getBody()));
        System.out.println("replyCode：" + replyCode);
        System.out.println("replyText：" + replyText);
        System.out.println("exchange：" + exchange);
        System.out.println("routingKey：" + routingKey);
    }

}
    消费端 
    消费端手动ACK确认

    实现如下：
    1.确认方法改为手动确认，配置文件：acknowledge-mode: manual

    server:
  port: 81
  
spring:
  rabbitmq:
    host: 192.168.31.6
    port: 5672
    username: admin
    password: admin
    virtual-host: /
    # 消费端手动ack确认队列消息
    listener:
      simple:
        acknowledge-mode: manual

    2.消费类增加Channel对象参数和deliveryTag参数
    3.channel.basicAck(deliveryTag, true)确认队列消息，channel.拒签，是否回到队列
    消息接收类具体代码示例
    Service

    public interface RabbitMqService {

    
    public void receiveSubMessageAck(String message, Channel channel, long deliveryTag);

}

    Service实现

    public class RabbitMqServiceImpl implements RabbitMqService {

    @Resource
    private AmqpTemplate amqpTemplate;

    @Override
    @RabbitListener(queues = {RabbitMQConfig.ACK_QUEUE})
    public void receiveSubMessageAck(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long deliveryTag) {
        try {
            System.out.println("接收到的mq消息：" + message);
            // 处理业务
            System.out.println("开始处理业务=========>");

            // 消费者手动确认队列消息(批量确认)
            if (deliveryTag == 5) {
                channel.basicAck(deliveryTag, true);// long deliveryTag 标号(当前消息在队列中的的索引)， boolean multiple 是否批量处理
            }
            // 消费者手动确认队列消息(单个确认，低效)
            // channel.basicAck(deliveryTag, true);
            
        } catch (Exception e) {
        	// 报错，消费者进行拒签
            e.printStackTrace();
            try {
                channel.basicNack(deliveryTag, false, false);//long deliveryTag 标号, boolean multiple 是否批量处理, boolean requeue 是否回到队列(false直接丢弃)
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }
}
  3. 怎么保证消息传递的顺序 
    开启多个队列，消息投递到同一个队列，一个消费者消费
  4. 一致性问题 
    一致性问题即事务问题，可使用nacos注册中心+seata解决分布式事务问题，下次更新

欢迎分享，转载请注明来源：内存溢出

原文地址: http://outofmemory.cn/zaji/5669633.html

赞 (0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
羚羊飞渡 羚羊飞渡 一级用户组
0 0
上一篇 2022-12-16
下一篇 2022-12-16

发表评论

登录后才能评论

评论列表（0条）

保存