rabbitmq的基本使用

亦凉 2022-03-07 03:28 176阅读 0赞

现在微服务盛行， 我们通常会进行解耦， 这时候就需要异步的消息队列来帮助各个服务之间解耦

## rabbitmq的基本概念介绍 ##

rabbitmq的基本概念有消息producer(消息生产者)、exchange(交换机)、queue(队列)、consumer(消费者)、routingKey

![clipboard.png][]  
(图中的P是producer， 即消息生产者， 中简的Server是Exchange(交换机) 和 Queue(队列))

*  **Queue**(队列)

queue是存放消息的队列， 实际上就是一个存放消息数据结构为队列的一个容器

*  **exchange**(交换机)

我们可能会简单的以为发送者会把消息发送到队列中， 然后消费者对队列进行监听。事实上， 消息发送者永远不会将消息直接发送到队列中， 而是将消息发送到exhang中， 再由exchange通过一定的路由规则路由到对应的消息队列中。  
交换机有四种类型:

Direct
    Topic
    Headers
    Fanout
    
    Direct Exchange:

![clipboard.png][clipboard.png 1]

*  **routingKey**

在上面介绍exchange中说到消息通过一定的路由规则路由到对应的队列中， routingKey就是起着这样的一个作用，通常我们发送消息到exchane中的时候会携带一个routingKey, 而这个routingKey就是exchange和queue绑定的一个规则， 由此便可以将消息从exchange再发送到对应的queue上

参考文章[https://segmentfault.com/a/11...][https_segmentfault.com_a_11...]

## SpringBoot中使用rabbitmq ##

首先添加以下依赖：

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

配置项如下：

spring:
      rabbitmq:
        port: 5672
        password: guest
        username: guest
        host: localhost
        listener:
          simple:
            acknowledge-mode: manual
            concurrency: 1
            max-concurrency: 1
            retry:
              enabled: true

在浏览器输入[http://localhost][http_localhost]:15672/, 在mq上我们新建了一个名为exchange1, routingKey为exhcange1-queue1的exchange， 并且映射到名为queue1的队列,

![图片描述][3969172599-5c667deeb675d_articlex]

发送消息代码Prodcuer：

public class Sender{
    
        @Autowired
        private RabbitTemplate rabbitTemplate;
    
        public void send(Object object) {
            CorrelationData correlationData = new CorrelationData();
            correlationData.setId("exchange1-queue1-id");
            String message = "hello world";
            rabbitTemplate.convertAndSend("exchange1", "exchange1-queue1", "helloworld", new CorrelationData());
        }
    }

在上面的代码中我们发送了一个消息到名为"exchange1", routingKey为"exchange1-queue1"的消息。我们启动rabbitmq。  
发送后可以在mq上看到如下图已经成功发送了。

![图片描述][1252805590-5c66831eb9b7e_articlex]

接下来贴上接受消息的receiver代码：

@Component
    public class Receiver {
    
        @RabbitListener(queues = "queue1")
        public void receive(Message message, Channel channel) {
            try {
                System.out.println(message.getBody());
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

接受消息后再看mq上如下图：  
![图片描述][2082041197-5c6684ed29aa5_articlex]

可以看出queue1上Ready一栏是0，但是Unacked一栏和Total一栏依然有消息， 这是因为我们再配置文件中设置的是手动的ack，这时候代码中没有进行ack, mq会认为消费者没有成功消费掉这条消息， 这时候就需要在配置文件中设置成自动ack, 或者在代码中手动进行ack,在消费者后添加如下代码：

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

[clipboard.png]: /images/20220307/7ff8f7a6f0a446bbbc9d256a94e40d65.png
[clipboard.png 1]: /images/20220307/82ec2d013f974b628807e649466824f6.png
[https_segmentfault.com_a_11...]: https://segmentfault.com/a/1190000016418132
[http_localhost]: http://localhost
[3969172599-5c667deeb675d_articlex]: /images/20220307/e8806ec2fc8a4ed8a4c29058e69889ea.png
[1252805590-5c66831eb9b7e_articlex]: /images/20220307/987d6c89eabc490e9ec04c575f96eed9.png
[2082041197-5c6684ed29aa5_articlex]: /images/20220307/b076532bc9f6454086b75788e181a87a.png