RabbitMQ的四种消息传递模式与演示代码

2024-04-09 104

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

本文涉及的产品

日志服务 SLS，月写入数据量 50GB 1个月

简介： RabbitMQ的四种消息传递模式与演示代码

RabbitMQ的四种消息传递模式与演示代码

RabbitMQ是一个功能强大的消息代理，提供了多种消息传递模式来满足不同场景下的需求。本文将介绍RabbitMQ的四种常用消息传递模式：Work、Fanout、Direct、Topic，并给出相应的Java示例代码。

1. Work模式

Work模式也被称为任务队列模式，它将任务分发给多个消费者，并由消费者竞争性地消费任务。每个任务只会被一个消费者处理。

Java示例代码：

import com.rabbitmq.client.*;

public class Worker {
    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] args) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        channel.basicQos(1);

        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 + "'");
                try {
                    doWork(message);
                } finally {
                    System.out.println(" [x] Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        channel.basicConsume(TASK_QUEUE_NAME, false, consumer);
    }

    private static void doWork(String task) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException _ignored) {
            Thread.currentThread().interrupt();
        }
    }
}

应用场景

一个常见的应用场景是在Web应用中实现异步任务处理。例如，用户在网站上提交了一个长时间处理的任务（如生成报表、发送邮件等），为了提高用户体验，可以将任务提交到RabbitMQ的任务队列中，然后由后台的消费者进行异步处理。这样一来，用户在网站上提交任务后即可立即得到响应，而不必等待任务处理完成，提高了系统的响应速度和并发处理能力。

下面是一个使用Work模式的简单示例代码，包括生产者和消费者部分，用Java编写，并添加了详细的注释。

生产者（Producer）代码：

import com.rabbitmq.client.*;

public class Producer {
    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] args) throws Exception {
        // 创建连接和频道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明队列
            channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
            
            // 构造消息
            String message = String.join(" ", args);
            
            // 发送消息
            channel.basicPublish("", TASK_QUEUE_NAME,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,
                    message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }
}

消费者（Consumer）代码：

import com.rabbitmq.client.*;

public class Consumer {
    private static final String TASK_QUEUE_NAME = "task_queue";

    public static void main(String[] args) throws Exception {
        // 创建连接和频道
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");
        final Connection connection = factory.newConnection();
        final Channel channel = connection.createChannel();

        // 声明队列
        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);
        System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

        // 设置消息处理的回调函数
        channel.basicConsume(TASK_QUEUE_NAME, false, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws java.io.IOException {
                String message = new String(body, "UTF-8");
                
                // 模拟任务处理过程
                try {
                    doWork(message);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    System.out.println(" [x] Done");
                    // 手动发送应答
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        });
    }

    private static void doWork(String task) throws InterruptedException {
        // 模拟任务处理过程，这里休眠1秒
        Thread.sleep(1000);
    }
}

在上述代码中，生产者（Producer）将消息发布到名为task_queue的队列中，而消费者（Consumer）则从该队列中获取任务并进行处理。每个任务都会被一个消费者处理，通过模拟任务处理过程，演示了Work模式的基本使用方式。

2. Fanout模式

Fanout模式是RabbitMQ中的一种消息传递模式，它将消息广播到所有绑定到Exchange的队列中，即使在消息发布之后才创建的队列，也能接收到消息。

原理解析

在Fanout模式中，生产者将消息发布到一个名为Exchange的交换机中，而不是直接发送到队列中。交换机会将收到的消息广播到与其绑定的所有队列中。因此，无论在消息发布之前还是之后创建的队列，只要它们与交换机进行了绑定，就能接收到交换机广播的消息。

在示例代码中，首先声明一个名为logs的Fanout类型的Exchange，并将消息发布到该Exchange中。消息发布时，并没有指定具体的队列，而是将消息发送到了Exchange中。Exchange会将消息广播到所有与其绑定的队列中，这就是Fanout模式的工作原理。

应用场景

一个常见的应用场景是日志处理系统。在一个分布式的日志系统中，通常会有多个日志消费者，它们分别负责处理不同级别（如info、error等）的日志。通过使用Fanout模式，可以将日志消息广播到所有相关的队列中，每个消费者只需要关注自己负责处理的日志级别，从而实现了日志的分发和处理。

通过Fanout模式，我们可以实现消息的广播传递，适用于多个消费者需要同时接收同一份消息的场景，例如日志处理系统、实时广播等。

Java示例代码：

import com.rabbitmq.client.*;

public class EmitLog {
    private static final String EXCHANGE_NAME = "logs";

    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.exchangeDeclare(EXCHANGE_NAME, "fanout");

            String message = getMessage(argv);

            channel.basicPublish(EXCHANGE_NAME, "", null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + message + "'");
        }
    }

    private static String getMessage(String[] strings) {
        if (strings.length < 1) {
            return "info: Hello World!";
        }
        return String.join(" ", strings);
    }
}

3. Direct模式

Direct模式将消息路由到与消息的RoutingKey完全匹配的队列中。

应用场景

一个常见的应用场景是日志级别过滤。在一个分布式的日志系统中，可能有多个消费者负责处理不同级别的日志（如info、error、warning等）。通过使用Direct模式，生产者可以根据日志的级别指定不同的RoutingKey，并将日志消息发布到Exchange中。Exchange会将日志消息路由到与其RoutingKey完全匹配的队列中，从而实现了日志的级别过滤和分发。

Java示例代码：

import com.rabbitmq.client.*;

public class EmitLogDirect {
    private static final String EXCHANGE_NAME = "direct_logs";

    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.exchangeDeclare(EXCHANGE_NAME, "direct");

            String severity = getSeverity(argv);
            String message = getMessage(argv);

            channel.basicPublish(EXCHANGE_NAME, severity, null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + severity + "':'" + message + "'");
        }
    }

    private static String getSeverity(String[] strings) {
        if (strings.length < 1) {
            return "info";
        }
        return strings[0];
    }

    private static String getMessage(String[] strings) {
        if (strings.length < 2) {
            return "Hello World!";
        }
        return String.join(" ", strings);
    }
}

4. Topic模式

Topic模式是RabbitMQ中的一种消息传递模式，它将消息发送到与匹配通配符的RoutingKey相匹配的队列中.

原理解析

在Topic模式中，生产者将消息发布到一个名为Exchange的交换机中，并且在发送消息时需要指定一个RoutingKey。RoutingKey可以使用通配符（和#）来匹配多个队列，其中表示匹配一个单词，#表示匹配零个或多个单词。交换机会将收到的消息根据RoutingKey和通配符匹配规则将消息路由到与之匹配的队列中。

在示例代码中，我们首先声明一个名为topic_logs的Topic类型的Exchange，并指定消息的RoutingKey。然后将消息发布到Exchange中，Exchange会根据消息的RoutingKey和通配符匹配规则将消息路由到与之匹配的队列中。

应用场景

一个常见的应用场景是日志过滤器。在一个分布式的日志系统中，可能有多个消费者负责处理不同模块或不同级别的日志。通过使用Topic模式，生产者可以根据日志的模块或级别指定特定的RoutingKey，并将日志消息发布到Exchange中。消费者可以使用通配符来订阅感兴趣的日志模块或级别，Exchange会将日志消息路由到与之匹配的队列中，从而实现了日志的模块化过滤和分发。

Java示例代码：

import com.rabbitmq.client.*;

public class EmitLogTopic {
    private static final String EXCHANGE_NAME = "topic_logs";

    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.exchangeDeclare(EXCHANGE_NAME, "topic");

            String routingKey = getRouting(argv);
            String message = getMessage(argv);

            channel.basicPublish(EXCHANGE_NAME, routingKey, null, message.getBytes("UTF-8"));
            System.out.println(" [x] Sent '" + routingKey + "':'" + message + "'");
        }
    }

    private static String getRouting(String[] strings) {
        if (strings.length < 1) {
            return "anonymous.info";
        }
        return strings[0];
    }

    private static String

 getMessage(String[] strings) {
        if (strings.length < 2) {
            return "Hello World!";
        }
        return String.join(" ", strings);
    }
}

RabbitMQ的四种消息传递模式与演示代码

RabbitMQ的四种消息传递模式与演示代码

1. Work模式

应用场景

2. Fanout模式

Java示例代码：

3. Direct模式

应用场景

Java示例代码：

4. Topic模式

原理解析

应用场景

热门文章

最新文章

相关课程

相关电子书

相关实验场景

热门

活动广场

任务中心

开发者评测

高校计划

乘风者计划

训练营

阿里云MVP

话题

直播

下载

镜像站

技术资料

插件

RabbitMQ的四种消息传递模式与演示代码

RabbitMQ的四种消息传递模式与演示代码

1. Work模式

应用场景

2. Fanout模式

Java示例代码：

3. Direct模式

应用场景

Java示例代码：

4. Topic模式

原理解析

应用场景

热门文章

最新文章

相关课程

相关电子书

相关实验场景