RabbitMQ学习-客户端API学习（springboot集成）

刺骨的言语ヽ痛彻心扉 2023-02-23 08:20 10阅读 0赞

## 写在前面 ##

这里介绍一下RabbitMQ客户端的操作方式

## 一、项目构建 ##

### 1.1、 引入依赖 ###

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>

### 1.2、yml 配置 ###

spring:
      rabbitmq:
        addresses: ip:5672
        username: root
        password: root
        virtual-host: /vhost
        # 开启confirms回调 P -> Exchange
        publisher-confirms: true
        # 开启returnedMessage回调 Exchange -> Queue
        publisher-returns: true
        # 设置手动确认(ack) Queue 
        listener:
          simple:
            acknowledge-mode: manual
        template:
          mandatory: true

### 1.3、producer ###

*  先定义回掉函数

//回调函数: confirm确认
        final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() { 
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) { 
                System.err.println("correlationData: " + correlationData);
                String messageId = correlationData.getId();
                if (ack) { 
                    //如果confirm返回成功
                } else { 
                    //失败则进行具体的后续操作:重试 或者补偿等手段
                    System.err.println("异常处理...");
                }
            }
        };

*  producer 1

public void sendOrder(Order order) throws Exception { 
            // 通过实现 ConfirmCallback 接口，消息发送到 Broker 后触发回调，确认消息是否到达 Broker 服务器，也就是只确认是否正确到达 Exchange 中
            rabbitTemplate.setConfirmCallback(confirmCallback);
            // 消息唯一ID
            CorrelationData correlationData = new CorrelationData(order.getMessageId());
            rabbitTemplate.convertAndSend("ls-exchange", "ls.ABC", order, correlationData);
        }

*  producer 2

public void sendOrder3(Order order) throws Exception { 
            rabbitTemplate.setConfirmCallback(confirmCallback);
            CorrelationData correlationData = new CorrelationData(order.getMessageId());
            rabbitTemplate.convertAndSend("ex1",
                    "ls.ABC",
                    order,
                    message -> { 
                        message.getMessageProperties().setPriority(1);
                        return message;
                    }, correlationData
            );
        }

### 1.4、consumer ###

//配置监听的哪一个队列，同时在没有queue和exchange的情况下会去创建并建立绑定关系
        @RabbitListener(bindings = @QueueBinding(
                value = @Queue(value = "ls-queue", durable = "true"),
                exchange = @Exchange(name = "ls-exchange", durable = "true", type = "topic"),
                key = "ls.*"))
        @RabbitHandler//如果有消息过来，在消费的时候调用这个方法
        public void onOrderMessage(@Payload Order order, @Headers Map<String, Object> headers, Channel channel) throws IOException { 
            logger.info("---收到消息，开始消费--，订单ID：{}，Header:{},channel:{}", order.getId(), headers.toString(), channel.toString());
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
     
            channel.basicAck(deliveryTag, false);
        }

/** * 不定义交换机,收发消息 * * @param order * @param headers * @param channel * @throws IOException */
        @RabbitListener(queues = "defaultExchangeQueue", concurrency = "5-40")
        public void onOrderMessage3(@Payload String order, @Headers Map<String, Object> headers, Channel channel) throws IOException { 
            //消费者操作
            logger.info("--不定义交换机，开始消费--，订单ID：{}，Header:{},channel:{}", order, headers.toString(), channel.toString());
            logger.info("重试次数：{}", headers.get("retry"));
            Long deliveryTag = (Long) headers.get(AmqpHeaders.DELIVERY_TAG);
            //ACK,确认一条消息已经被消费
            channel.basicAck(deliveryTag, false);
        }