消息队列消费模型争霸！看懂秒会！

2023-12-04 04:08:34

探索消息队列的消费模型：揭开信息传递的秘密

队列模型：可靠且有序的消息传递

队列模型是最基本的消费模型，以其可靠性和顺序性而著称。在此模型中，每个消息都会被一个且仅一个消费者处理，保证了消息的完整传递。就像一个有序的队列，消息依次排队，确保它们按顺序被处理。这种模型非常适合需要确保数据完整性和处理顺序的场景，例如交易处理和库存管理。

代码示例：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.MessageConsumer;
import java.io.IOException;

public class QueueConsumer {

    public static void main(String[] args) throws IOException {
        // 连接到消息队列
        ConnectionFactory factory = new ConnectionFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明队列
        String queueName = "my-queue";
        channel.queueDeclare(queueName, false, false, false, null);

        // 创建消费者
        MessageConsumer consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                String message = new String(body);
                System.out.println("Received message: " + message);
            }
        };

        // 消费队列消息
        channel.basicConsume(queueName, true, consumer);
    }
}

发布订阅模型：多对多的灵活通信

发布订阅模型打破了队列模型的单一消费限制。在这个模型中，一个消息可以被多个消费者同时处理。类似于订阅报纸，多个订阅者可以接收同一份消息。这种模型非常适合需要广泛传播消息的场景，例如新闻更新和社交媒体通知。

代码示例：

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.DeliverCallback;
import java.io.IOException;

public class TopicConsumer {

    public static void main(String[] args) throws IOException {
        // 连接到消息队列
        ConnectionFactory factory = new ConnectionFactory();
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        // 声明交换器
        String exchangeName = "my-topic-exchange";
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC, false, false, null);

        // 声明队列并绑定到交换器
        String queueName = "my-topic-queue";
        channel.queueDeclare(queueName, false, false, false, null);
        channel.queueBind(queueName, exchangeName, "#");

        // 创建消费者
        DeliverCallback callback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody());
            System.out.println("Received message: " + message);
        };

        // 消费队列消息
        channel.basicConsume(queueName, true, callback, consumerTag -> {});
    }
}