当前位置：首页 > article >正文

RabbitMQ---事务及消息分发

article 2025/1/21 8:54:10

（一）事务

RabbitMQ是基于AMQP协议实现的，该协议实现了事务机制，所以RabbitMQ也支持事务机制，他的事务允许开发者确保消息的发送和接收时原子性的，要么全部成功，要么全部失败

我们设置事务有三步，首先就是开启事务

因为我们是针对rabbitTemplate来作设置的，所以会影响此rabbitTemplate的所有消息，这里我们新开了一个

然后我们使用时要加一个注解

最后一步我们需要加上事务管理

这样我们就成功开启了事务

这三步是必不可少的，缺少一步，都无法成功开启事务

(二)消息分发

1.概念

RabbitMQ队列有多个消费者时，队列会把消息分给不同的消费者，每条消息只发给一个消费者，默认情况下，RabbitMQ是按照轮询的方式分发的，不管消费者是否已经消费并确认消息，所以这种方式是很有可能造成消息积压的，因为有的消费者处理消息的速度快，有的就慢，这样会导致整体吞吐量下降

那至于我们要如何处理，我们可以通过channelbasic（）这个方法来限制信道上的最大消息数量，RabbitMQ每向该消费者发送消息，就会使消息计数+1，消费者消费消息就会使消息计数-1，当到达了上线，RabbitMQ就不会再发送消息了，知道消费者又消费了一条消息

2.应用场景

我们可以将消息分发应用在限流.非公平分发（负载均衡中）

1）限流

我们来看一个例子，我们有一个订单系统，每秒可以处理5000个请求，在正常情况下，是不会有问题的，但是在一些特殊时间比如11.11，请求量就会突增，如果这些请求全部直接发送到消费者（订单系统），那么就会把我们订单系统给弄崩了

RabbitMQ提供了限流机制，可以控制消费者一次最多拉取多少个请求，通过在配置文件中设置prefetch参数即可

我们来看代码

因为要限流，限制拉取的请求，所以我们这里要设置手动确认，如果我们设置成自动确认，那我们收到消息就确认，其实跟没有设置限流差别是不大的

还是先看配置代码

spring:
  rabbitmq:
    addresses: amqp://student:student@62.234.46.219:5672/test
    listener:
      simple:
        #        acknowledge-mode: NONE
        #        acknowledge-mode: AUTO
        acknowledge-mode: MANUAL
        prefetch: 5
        retry:
          enabled: true # 开启消费者失败重试
          initial-interval: 5000ms # 初始失败等待时⻓为5秒
          max-attempts: 5
#    publisher-confirm-type: correlated  #消息发送确认

这里我们设置一次最多拉取5条

然后我们生产者一次发送10条消息

@RestController
@RequestMapping("producer")
public class Qos {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @RequestMapping("/qos")
    public String qosConsumer(){
        for (int i = 0; i < 10; i++) {
            rabbitTemplate.convertAndSend(Constant.QOS_EXCHANGE,"qos","qos "+i);
        }
        return "发送成功";
    }
}

然后消费者消费消息，但是这里我们没有进行确认，主要是为了方便观察结果，如果我们确认了，本身我们消息就不多而且代码中没什么复杂逻辑，处理的很快，肉眼上就像一次全部获取了一样

@Component
public class QosConsumer {
    @RabbitListener(queues = Constant.QOS_QUEUE)
    public void listenerQueue(Message message, Channel channel) throws UnsupportedEncodingException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        System.out.printf("接收到消息: %s, deliveryTag: %d%n",
                new String(message.getBody(),"UTF-8"), deliveryTag);
    }
}

之后我们来看结果

接收到了5条消息

然后我们看看控制台，发现确实有5条还没有发送，有5条已经确认了

那我们再把刚刚配置文件的限制取消了，再看看结果

我们发现消息会直接全部发送到消费者

2）负载均衡

我们也可以用这个配置来实现负载均衡

这里我们说的负载均衡，并不是说两个消费者每个人都处理相同数量的消息，而是两个人谁干的快谁就多干，谁干的慢谁就少干

如图，我们有两个消费者，一个处理的快，一个处理的满，就会有消费者快的处理完后不知道干什么，而另一个消费者还一直在处理，这是因为RabbitMQ只是在消息进入队列时分配消息，并不考虑消费者未确认消息的数量

我们可以设置prefetch为1，这样每个消费者都会只有一条消息，在这条消息处理完之前，就不会有其他消息

对于消费者来说就是，干的越快的干的越多

那其实我个人认为限流和负载均衡是很像的，因为负载均衡就是多个消费者一起限流，然后谁干的快，谁就再拿，而且上述例子上的prdfetch为1也不一定，我们也可以是其他值

接下来我们看代码

这里我们只需要多加一个消费者然后把配置文件prefetch改为1即可

@Component
public class QosConsumer {
    @RabbitListener(queues = Constant.QOS_QUEUE)
    public void listenerQueue(Message message, Channel channel) throws IOException, InterruptedException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        Thread.sleep(1000);
        System.out.printf("1接收到消息: %s, deliveryTag: %d%n",
                new String(message.getBody(),"UTF-8"), deliveryTag);
        channel.basicAck(deliveryTag,false);
    }
    @RabbitListener(queues = Constant.QOS_QUEUE)
    public void listenerQueue2(Message message, Channel channel) throws IOException, InterruptedException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        Thread.sleep(2000);
        System.out.printf("2接收到消息: %s, deliveryTag: %d%n",
                new String(message.getBody(),"UTF-8"), deliveryTag);
        channel.basicAck(deliveryTag,false);
    }
}

然后我们看现象，我们发现只有我们确认了之后，消息才会在再发送给这个消费者，来达到负载均衡的效果