RabbitMQ在项目中做什么用？怎么消费消息？具体怎么使用的？

程序员朱永胜

发布于 2024-01-07 10:39:09

5470

发布于 2024-01-07 10:39:09

文章被收录于专栏：朱永胜的私房菜

有的时候博客内容会有变动，首发博客是最新的，其他博客地址可能会未同步,认准https://blog.zysicyj.top

RabbitMQ 在项目中的用途

RabbitMQ 是一个开源的消息代理和队列服务器，用于在分布式系统之间异步传递消息。它使用AMQP（高级消息队列协议）来传输消息，并支持多种消息传输模式。

在项目中，RabbitMQ 的几个主要用途如下：

「1. 解耦应用组件」

通过使用消息队列，生产者（发送消息的应用程序）和消费者（接收消息的应用程序）可以独立运行和扩展，它们之间不需要直接通信，从而达到解耦的目的。

「2. 异步处理」

RabbitMQ 允许应用程序将任务发送到队列中，而不是直接进行处理。这样可以让用户请求快速返回，提高系统的响应性能，而实际的任务处理可以异步进行。

「3. 流量削峰」

在高峰时段，RabbitMQ 可以帮助系统缓存过多的请求，平滑处理压力高峰，当流量减少时再逐渐处理这些请求。

「4. 可靠性保证」

RabbitMQ 支持消息持久化，确保在服务器崩溃的情况下，消息不会丢失，从而提高系统的可靠性。

消费消息的方式

消息的消费通常指的是应用程序从RabbitMQ队列中取出并处理消息的过程。以下是消费消息的基本步骤：

「1. 建立连接」

首先，消费者应用程序需要与RabbitMQ 服务器建立连接。

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("your.rabbitmq.host");
Connection connection = factory.newConnection();
Channel channel = connection.createChannel();

「2. 声明队列」

确保队列存在，如果不存在将会创建它。

channel.queueDeclare("queue_name", false, false, false, null);

「3. 创建消费者」

创建消费者，并告诉它如何处理消息。

Consumer consumer = new DefaultConsumer(channel) {
    @Override
    public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
        String message = new String(body, "UTF-8");
        System.out.println(" [x] Received '" + message + "'");
        // 这里进行消息处理逻辑
    }
};

「4. 开始消费消息」

让消费者开始监听队列，并消费消息。

channel.basicConsume("queue_name", true, consumer);

在这个基本的消费过程中，消费者通过实现 DefaultConsumer 类并重写 handleDelivery 方法来处理接收到的消息。"queue_name" 是指定的队列名，消费者将监听这个队列获取消息。

具体的使用方式

以下是一个简单的例子，展示如何在Java项目中使用RabbitMQ:

import com.rabbitmq.client.*;

public class Recv {
    private final static String QUEUE_NAME = "hello";

    public static void main(String[] argv) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(QUEUE_NAME, false, false, false, null);
            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                String message = new String(delivery.getBody(), "UTF-8");
                System.out.println(" [x] Received '" + message + "'");
            };
            channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> { });
        }
    }
}

在这个例子中，我们创建了一个消费者来异步接收队列hello中的消息。每当接收到消息，就会打印出来。

这只是RabbitMQ的基本使用。实际项目中，你还可能需要处理连接的恢复、消息的确认、交换器的使用、队列的绑定、死信处理、消息的持久化等高级功能。

本文由 mdnice 多平台发布

本文参与腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。

原始发表：2024-01-06，如有侵权请联系 cloudcommunity@tencent.com 删除

rabbitmq