美文网首页
RabbitMq: 使用

RabbitMq: 使用

作者: 布拉德老瓜 | 来源:发表于2021-10-08 23:07 被阅读0次

    1. 创建队列

        public Queue(String name, boolean durable, boolean exclusive, boolean autoDelete, @Nullable Map<String, Object> arguments) {
            super(arguments);
            Assert.notNull(name, "'name' cannot be null");
            this.name = name;
            this.actualName = StringUtils.hasText(name) ? name : Base64UrlNamingStrategy.DEFAULT.generateName() + "_awaiting_declaration";
            this.durable = durable; // 是否持久化
            this.exclusive = exclusive; // 队列中消息是否只提供一个消费者消费
            this.autoDelete = autoDelete; // 当全部消费者断开连接后,队列中消息是否自动删除
        }
    

    2. 创建exchange

    public TopicExchange(String name, boolean durable, boolean autoDelete, Map<String, Object> arguments) {
        super(name, durable, autoDelete, arguments);
    }
    

    3. 绑定exchange与队列

        public Binding(String destination, Binding.DestinationType destinationType, String exchange, String routingKey, @Nullable Map<String, Object> arguments) {
            super(arguments);
            this.destination = destination;
            this.destinationType = destinationType;
            this.exchange = exchange;
            this.routingKey = routingKey;
        }
    

    Binding.DestinationType可以是:QUEUE, 或EXCHANGE
    destination为目的地,它可以是一个队列或者是exchange
    exchange为数据发送的源exchange
    routeKey为匹配投递的路由key.

    4.消息发送

    生产者可以使用rabbitTemplate通过以下方式发送消息:

    • 4.1. 将object序列化为json或其他格式(默认采用的是JAVA的序列化),直接发送到指定exchange,通过routeKey转发到相应的队列.
        public void convertAndSend(String exchange, String routingKey, Object object, @Nullable CorrelationData correlationData) throws AmqpException {
            this.send(exchange, routingKey, this.convertMessageIfNecessary(object), correlationData);
        }
    
    • 4.2 发送前先通过postProcessor完成消息的处理,然后将处理后的消息发送.
        public void convertAndSend(String exchange, String routingKey, Object message, MessagePostProcessor messagePostProcessor, @Nullable CorrelationData correlationData) throws AmqpException {
            Message messageToSend = this.convertMessageIfNecessary(message);
            messageToSend = messagePostProcessor.postProcessMessage(messageToSend, correlationData);
            this.send(exchange, routingKey, messageToSend, correlationData);
        }
    

    5. 消息消费

    • 5.1 消费者监听队列
      通过@RabbitListener明一个消费者,并声明他监听的队列.通过@RabbitHandler声明消息被接收后的处理器.同一个消费者可以声明多个处理器,用来处理不同类型的消息内容实体.
    @RabbitListener(queues = {"packet-out-of-ware-queue"})
    public class OrderConsumer {
        @RabbitHandler
        public void receiveMsg(Message message, OrderEntity orderEntity){
            System.out.println(message + "\n" + orderEntity);
        }
    
        // 处理来自于packet-out-of-ware队列中,消息体为OrderRefundReson的消息
        @RabbitHandler
        public void receiveMsg(Message message, OrderRefundReson orderRefundReson){
            System.out.println(message + "\n" + orderEntity);
        }
    }
    

    6. 消息可靠投递

    6.1 生产者发送,broker确认

    包含两个阶段确认:

      1. 生产者消息发送到broker, broker收到后返回确认信息,需配置
    spring.rabbitmq..publisher-confirm-type: simple  (或 correlated)
    
      1. 生产者消息发送到broker并由exchange将消息发送到指定队列, 如果没能找到队列,返回失败信息,需配置
        spring.rabbitmq.publisher-returns: true
        spring.rabbitmq.template.mandatory: true
    
        @PostConstruct
        public void initRabbitTemplate(){
            rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
                /**
                 * 档消息被投递到exchange,发送确认后,触发回调
                 * @param correlationData:关联信息,如消息id 
                 * @param ack
                 * @param cause
                 */
                @Override
                public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                    System.out.println("confirming... " + "correlationDataId: " + correlationData.getId() + "ack: " + ack);
                }
            });
    
            rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
                /**
                 * 当消息未被投递到指定队列,如根据routeKey查找队列不存在则触发returnedMessage
                 * @param message
                 * @param replyCode
                 * @param replyText
                 * @param exchange
                 * @param routingKey
                 */
                @Override
                public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
    
                }
            });
        }
    

    6.2 消费者确认

    默认为自动确认,可以选择通过与手动确认

        @RabbitHandler
        public void receiveMsg(Message message, OrderEntity orderEntity, Channel channel){
            try {
                System.out.println(message + "\n" + orderEntity);
                boolean done = handleMsg(orderEntity);
                long deliveryTag = message.getMessageProperties().getDeliveryTag();
                if (done){
                    // 处理成功,以非批量的方式返回确认信息
                    // basicNack(deliveryTag, multiple)
                    channel.basicAck(deliveryTag, false);
                }else {
         // 处理失败,以非批量的方式返回Nack信息,同时可选择是否将消息重新放回队列. 
                   //basicNack(deliveryTag, multiple, requeue)
                    channel.basicNack(deliveryTag, false, false);
                }
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
    
    

    相关文章

      网友评论

          本文标题:RabbitMq: 使用

          本文链接:https://www.haomeiwen.com/subject/waqfoltx.html