RabbitMQ的使用

2023-11-19

安装：

Docker 安装 RabbitMQ

docker run -d --name rabbitmq -p 5671:5671 -p 5672:5672 -p 4369:4369 -p  25672:25672 -p 15671:15671 -p 15672:15672 rabbitmq:management

4369,25672(Erlang发现&集群端口)

5672,5671(AMQP端口)

15672 (web管理后台端口)

61613,61614(STOMP协议端口)

1883,8883(MQTT协议端口)

RabbitMQ随docker自动启动

docker update rabbitmq --restart=always

配置

设置传送rabbitmq格式化，

@Configuration
public class MyRabbitConfig {

    private RabbitTemplate rabbitTemplate;

    @Primary
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        this.rabbitTemplate = rabbitTemplate;
        rabbitTemplate.setMessageConverter(messageConverter());
        initRabbitTemplate();
        return rabbitTemplate;
    }

    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    /**
     * 定制RabbitTemplate
     * 1、服务收到消息就会回调
     *      1、spring.rabbitmq.publisher-confirms: true
     *      2、设置确认回调
     * 2、消息正确抵达队列就会进行回调
     *      1、spring.rabbitmq.publisher-returns: true
     *         spring.rabbitmq.template.mandatory: true
     *      2、设置确认回调ReturnCallback
     *
     * 3、消费端确认(保证每个消息都被正确消费，此时才可以broker删除这个消息)
     *
     */
    // @PostConstruct  //MyRabbitConfig对象创建完成以后，执行这个方法
    public void initRabbitTemplate() {

        /**
         * 1、只要消息抵达Broker就ack=true
         * correlationData：当前消息的唯一关联数据(这个是消息的唯一id)
         * ack：消息是否成功收到
         * cause：失败的原因
         */
        //设置确认回调
        rabbitTemplate.setConfirmCallback((correlationData,ack,cause) -> {
            System.out.println("confirm...correlationData["+correlationData+"]==>ack:["+ack+"]==>cause:["+cause+"]");
        });


        /**
         * 只要消息没有投递给指定的队列，就触发这个失败回调
         * message：投递失败的消息详细信息
         * replyCode：回复的状态码
         * replyText：回复的文本内容
         * exchange：当时这个消息发给哪个交换机
         * routingKey：当时这个消息用哪个路邮键
         */
        rabbitTemplate.setReturnCallback((message,replyCode,replyText,exchange,routingKey) -> {
            System.out.println("Fail Message["+message+"]==>replyCode["+replyCode+"]" +
                    "==>replyText["+replyText+"]==>exchange["+exchange+"]==>routingKey["+routingKey+"]");
        });
    }
}

解决消息丢失

1、消息丢失

消息发送出去，由于网络问题没有抵达服务器

做好容错方法（try-catch），发送消息可能会网络失败，失败后要有重试机制，可记录到数据库，采用定期扫描重发的方式

做好日志记录，每个消息状态是否都被服务器收到都应该记录 • 做好定期重发，如果消息没有发送成功，定期去数据库扫描未成功的消息进行重发

消息抵达Broker，Broker要将消息写入磁盘（持久化）才算成功。此时Broker尚未持久化完成，宕机。

publisher也必须加入确认回调机制，确认成功的消息，修改数据库消息状态。

自动ACK的状态下。消费者收到消息，但没来得及消息然后宕机

一定开启手动ACK，消费成功才移除，失败或者没来得及处理就noAck并重新入队

代码方法：

# 开启发送端消息抵达Broker确认
spring.rabbitmq.publisher-confirms=true
# 开启发送端消息抵达Queue确认
spring.rabbitmq.publisher-returns=true
# 只要消息抵达Queue，就会异步发送优先回调returnfirm
spring.rabbitmq.template.mandatory=true
# 手动ack消息，不使用默认的消费端确认
spring.rabbitmq.listener.simple.acknowledge-mode=manual

手动接受代码basicAck

channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);

ture为接受后删除，false为接受后不删除

拒收basicReject

 channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);

ture为拒收后重新入队，false为拒收后丢弃

解决消息重复：

• 2、消息重复 • 消息消费成功，事务已经提交，ack时，机器宕机。导致没有ack成功，Broker的消息重新由unack变为ready，并发送给其他消费者

• 消息消费失败，由于重试机制，自动又将消息发送出去

• 成功消费，ack时宕机，消息由unack变为ready，Broker又重新发送

消费者的业务消费接口应该设计为幂等性的。比如扣库存有工作单的状态标志

• 使用防重表（redis/mysql），发送消息每一个都有业务的唯一标识，处理过就不用处理 •rabbitMQ的每一个消息都有redelivered字段，可以获取是否是被重新投递过来的，而不是第一次投递过来的

解决消息积压：

• 3、消息积压 • 消费者宕机积压 • 消费者消费能力不足积压 • 发送者发送流量太大

• 上线更多的消费者，进行正常消费

• 上线专门的队列消费服务，将消息先批量取出来，记录数据库，离线慢慢处

简单使用

设计队列流程图

只要使用@bean注解交给spring管理即可，使用全量构造方法创建

@Configuration
public class MyRabbitMQConfig {

    /* 容器中的Queue、Exchange、Binding 会自动创建（在RabbitMQ）不存在的情况下 */

    /**
     * 死信队列
     *
     * @return
     */@Bean
    public Queue orderDelayQueue() {
        /*
            Queue(String name,  队列名字
            boolean durable,  是否持久化
            boolean exclusive,  是否排他
            boolean autoDelete, 是否自动删除
            Map<String, Object> arguments) 属性
         */
        HashMap<String, Object> arguments = new HashMap<>();
        arguments.put("x-dead-letter-exchange", "order-event-exchange");
        arguments.put("x-dead-letter-routing-key", "order.release.order");
        arguments.put("x-message-ttl", 60000); // 消息过期时间 1分钟
        Queue queue = new Queue("order.delay.queue", true, false, false, arguments);

        return queue;
    }

    /**
     * 普通队列
     *
     * @return
     */
    @Bean
    public Queue orderReleaseQueue() {

        Queue queue = new Queue("order.release.order.queue", true, false, false);

        return queue;
    }

    /**
     * TopicExchange
     *
     * @return
     */
    @Bean
    public Exchange orderEventExchange() {
        /*
         *   String name,
         *   boolean durable,
         *   boolean autoDelete,
         *   Map<String, Object> arguments
         * */
        return new TopicExchange("order-event-exchange", true, false);

    }


    @Bean
    public Binding orderCreateBinding() {
        /*
         * String destination, 目的地（队列名或者交换机名字）
         * DestinationType destinationType, 目的地类型（Queue、Exhcange）
         * String exchange,
         * String routingKey,
         * Map<String, Object> arguments
         * */
        return new Binding("order.delay.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.create.order",
                null);
    }

    @Bean
    public Binding orderReleaseBinding() {

        return new Binding("order.release.order.queue",
                Binding.DestinationType.QUEUE,
                "order-event-exchange",
                "order.release.order",
                null);
    }

}

监听者

主类上添加注解

@RabbitListener(queues = "order.release.order.queue")

方法上添加注解

@RabbitHandler
public void listener(OrderEntity orderEntity, Channel channel, Message message)

只需要将参数设置号rabbitMQ就会自动传参

发送消息

rabbitTemplate.convertAndSend("order-event-exchange","order.create.order",order.getOrder());

死性队列

安装

1. 首先我们将刚下载下来的
rabbitmq_delayed_message_exchange-3.8.0.ez文件上传到RabbitMQ所在服务器
2. 切换到插件所在目录，
执行 docker cp rabbitmq_delayed_message_exchange-3.8.0.ez rabbitmq:/plugins 命令，将刚插件拷贝到容器内plugins目录下
3. 执行 docker exec -it rabbitmq /bin/bash 
命令进入到容器内部，并 cd plugins 进入plugins目录
4. 执行 ls -l|grep delay  命令查看插件是否copy成功
5. 在容器内plugins目录下，执行 rabbitmq-plugins enable rabbitmq_delayed_message_exchange  命令启用插件
6. exit命令退出RabbitMQ容器内部，然后执行 docker restart rabbitmq 命令重启RabbitMQ容器

下载地址:Releases · rabbitmq/rabbitmq-delayed-message-exchange · GitHub

使用版本：

出现x-delayed-message交换机视为安装成功

本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)

Docker

Linux

Java