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

2024-04-09 64

版权

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

本文涉及的产品

日志服务 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模式

原理解析

应用场景

热门文章

最新文章

相关课程

相关电子书

相关实验场景