RabbitMQ的5种消息队列

rabbitmq的5种消息队列

1、七种模式介绍与应用场景

1.1 简单模式(hello world)

一个生产者对应一个消费者，rabbitmq 相当于一个消息代理，负责将 a 的消息转发给 b。

应用场景：将发送的电子邮件放到消息队列，然后邮件服务在队列中获取邮件并发送给收件人。

1.2 工作队列模式(work queues)

在多个消费者之间分配任务(竞争的消费者模式)，一个生产者对应多个消费者，一般适用于执行资源密集型任务，

单个消费者处理不过来，需要多个消费者进行处理。

应用场景：一个订单的处理需要10s，有多个订单可以同时放到消息队列，然后让多个消费者同时处理，这样就是

并行了，而不是单个消费者的串行情况。

1.3 订阅模式(publish/subscribe)

一次向许多消费者发送消息，一个生产者发送的消息会被多个消费者获取，也就是将消息将广播到所有的消费者

中。

应用场景：更新商品库存后需要通知多个缓存和多个数据库，这里的结构应该是：

• 一个 fanout 类型交换机扇出两个个消息队列，分别为缓存消息队列、数据库消息队列
• 一个缓存消息队列对应着多个缓存消费者
• 一个数据库消息队列对应着多个数据库消费者

1.4 路由模式(routing)

有选择地 (routing key) 接收消息，发送消息到交换机并且要指定路由 key，消费者将队列绑定到交换机时需要指

定路由 key，仅消费指定路由 key 的消息。

应用场景：如在商品库存中增加了 1 台 iphone12，iphone12 促销活动消费者指定 routing key 为 iphone12，只

有此促销活动会接收到消息，其它促销活动不关心也不会消费此 routing key 的消息。

1.5 主题模式(topics)

根据主题 (topics) 来接收消息，将路由 key 和某模式进行匹配，此时队列需要绑定在一个模式上，#匹配一个词

或多个词，*只匹配一个词。

应用场景：同上 iphone 促销活动可以接收主题为 iphone 的消息，如 iphone12、iphone13 等。

1.6 远程过程调用(rpc)

如果我们需要在远程计算机上运行功能并等待结果就可以使用 rpc，具体流程可以看图。

应用场景：需要等待接口返回数据，如订单支付。

1.7 发布者确认(publisher confirms)

与发布者进行可靠的发布确认，发布者确认是 rabbitmq 扩展，可以实现可靠的发布。在通道上启用发布者确认

后，rabbitmq 将异步确认发送者发布的消息，这意味着它们已在服务器端处理。

应用场景：对于消息可靠性要求较高，比如钱包扣款。

2、代码演示

<dependency>
	<groupid>com.rabbitmq</groupid>
	<artifactid>amqp-client</artifactid>
	<version>5.6.0</version>
</dependency>

2.1 简单模式

package simple;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class sender {

    private final static string queue_name = "simple_queue";

    public static void main(string[] args) throws ioexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        // 声明队列
        // queue: 队列名
        // durable: 是否持久化
        // exclusive: 是否排外,即只允许该channel访问该队列,一般等于true的话用于一个队列只能有一个消费者来消费的场景
        // autodelete: 是否自动删除,消费完删除
        // arguments: 其他属性
        channel.queuedeclare(queue_name, false, false, false, null);
        // 消息内容
        string message = "simplest mode message";
        channel.basicpublish("", queue_name, null, message.getbytes());
        system.out.println("[x]sent '" + message + "'");
        // 最后关闭通关和连接
        channel.close();
        connection.close();
    }
}

package simple;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class receiver {

    private final static string queue_name = "simple_queue";

    public static void main(string[] args) throws ioexception, timeoutexception {
        // 获取连接
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.queuedeclare(queue_name, false, false, false, null);
        delivercallback delivercallback = (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), "utf-8");
            system.out.println(" [x] received '" + delivery.getenvelope().getroutingkey() + "':'" + message + "'");
        };
        channel.basicconsume(queue_name, true, delivercallback, consumertag -> {
        });
    }
}

[x]sent 'simplest mode message'

[x] received 'simple_queue':'simplest mode message'

2.2 工作队列模式

package work;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class receiver1 {

    private final static string queue_name = "queue_work";

    public static void main(string[] args) throws ioexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.queuedeclare(queue_name, false, false, false, null);
        // 同一时刻服务器只会发送一条消息给消费者
        channel.basicqos(1);
        delivercallback delivercallback = (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), "utf-8");
            system.out.println(" [x-1] received '" + delivery.getenvelope().getroutingkey() + "':'" + message + "'");
        };
        channel.basicconsume(queue_name, true, delivercallback, consumertag -> {
        });
    }
}

package work;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class receiver2 {

    private final static string queue_name = "queue_work";

    public static void main(string[] args) throws ioexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.queuedeclare(queue_name, false, false, false, null);
        // 同一时刻服务器只会发送一条消息给消费者
        channel.basicqos(1);
        delivercallback delivercallback = (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), "utf-8");
            system.out.println(" [x-2] received '" + delivery.getenvelope().getroutingkey() + "':'" + message + "'");
        };
        channel.basicconsume(queue_name, true, delivercallback, consumertag -> {
        });
    }
}

package work;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class sender {

    private final static string queue_name = "queue_work";

    public static void main(string[] args) throws ioexception, interruptedexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        // 声明队列
        channel.queuedeclare(queue_name, false, false, false, null);
        for (int i = 0; i < 10; i++) {
            string message = "work mode message" + i;
            channel.basicpublish("", queue_name, null, message.getbytes());
            system.out.println("[x] sent '" + message + "'");
            thread.sleep(i * 10);
        }
        channel.close();
        connection.close();
    }
}

[x] sent 'work mode message0'
[x] sent 'work mode message1'
[x] sent 'work mode message2'
[x] sent 'work mode message3'
[x] sent 'work mode message4'
[x] sent 'work mode message5'
[x] sent 'work mode message6'
[x] sent 'work mode message7'
[x] sent 'work mode message8'
[x] sent 'work mode message9'

[x-1] received 'queue_work':'work mode message0'
[x-1] received 'queue_work':'work mode message2'
[x-1] received 'queue_work':'work mode message4'
[x-1] received 'queue_work':'work mode message6'
[x-1] received 'queue_work':'work mode message8'
 
[x-2] received 'queue_work':'work mode message1'
[x-2] received 'queue_work':'work mode message3'
[x-2] received 'queue_work':'work mode message5'
[x-2] received 'queue_work':'work mode message7'
[x-2] received 'queue_work':'work mode message9'

2.3 发布订阅模式

package publishsubscribe;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

public class receive1 {

    private static final string exchange_name = "logs";

    public static void main(string[] argv) throws exception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.exchangedeclare(exchange_name, "fanout");
        string queuename = channel.queuedeclare().getqueue();
        channel.queuebind(queuename, exchange_name, "");
        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-1] received '" + message + "'");
        };
        // 消费者，有消息时出发订阅回调函数
        channel.basicconsume(queuename, true, delivercallback, consumertag -> {
        });
    }
}

package publishsubscribe;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

public class receive2 {

    private static final string exchange_name = "logs";

    public static void main(string[] argv) throws exception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.exchangedeclare(exchange_name, "fanout");
        string queuename = channel.queuedeclare().getqueue();
        channel.queuebind(queuename, exchange_name, "");
        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-2] received2 '" + message + "'");
        };
        // 消费者，有消息时出发订阅回调函数
        channel.basicconsume(queuename, true, delivercallback, consumertag -> {
        });
    }
}

package publishsubscribe;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;

public class sender {

    private static final string exchange_name = "logs";

    public static void main(string[] argv) throws exception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.exchangedeclare(exchange_name, "fanout");
        string message = "publish subscribe message";
        channel.basicpublish(exchange_name, "", null, message.getbytes("utf-8"));
        system.out.println(" [x] sent '" + message + "'");
        channel.close();
        connection.close();
    }
}

[x] sent 'publish subscribe message'

[*] waiting for messages. to exit press ctrl+c
[x-1] received 'publish subscribe message'
 
[*] waiting for messages. to exit press ctrl+c
[x-2] received2 'publish subscribe message'

2.4 路由模式

package router;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class receiver1 {

    private final static string queue_name = "queue_routing";
    private final static string exchange_name = "exchange_direct";

    public static void main(string[] args) throws ioexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.queuedeclare(queue_name, false, false, false, null);
        // 指定路由的key,接收key和key2
        channel.queuebind(queue_name, exchange_name, "key");
        channel.queuebind(queue_name, exchange_name, "key2");
        channel.basicqos(1);
        delivercallback delivercallback = (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), "utf-8");
            system.out.println(" [x-1] received '" + delivery.getenvelope().getroutingkey() + "':'" + message + "'");
        };
        channel.basicconsume(queue_name, true, delivercallback, consumertag -> {
        });
    }
}

package router;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;


public class receiver2 {

    private final static string queue_name = "queue_routing2";
    private final static string exchange_name = "exchange_direct";

    public static void main(string[] args) throws ioexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.queuedeclare(queue_name, false, false, false, null);
        // 仅接收key2
        channel.queuebind(queue_name, exchange_name, "key2");
        channel.basicqos(1);
        delivercallback delivercallback = (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), "utf-8");
            system.out.println(" [x-2] received '" +
                    delivery.getenvelope().getroutingkey() + "':'" + message + "'");
        };
        channel.basicconsume(queue_name, true, delivercallback, consumertag -> {
        });
    }
}

package router;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;


public class sender {

    private final static string exchange_name = "exchange_direct";
    private final static string exchange_type = "direct";

    public static void main(string[] args) throws ioexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        // 交换机声明
        channel.exchangedeclare(exchange_name, exchange_type);
        // 只有routingkey相同的才会消费
        string message = "routing mode message";
        channel.basicpublish(exchange_name, "key2", null, message.getbytes());
        system.out.println("[x] sent '" + message + "'");
        // channel.basicpublish(exchange_name, "key", null, message.getbytes());
        // system.out.println("[x] sent '" + message + "'");
        channel.close();
        connection.close();
    }
}

[x] sent 'routing mode message'

[x-1] received 'key2':'routing mode message'

[x-2] received 'key2':'routing mode message'

2.5 主题模式

package topic;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class receiver1 {

    private final static string queue_name = "queue_topic";
    private final static string exchange_name = "exchange_topic";

    public static void main(string[] args) throws ioexception, interruptedexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.queuedeclare(queue_name, false, false, false, null);
        // 可以接收key.1
        channel.queuebind(queue_name, exchange_name, "key.*");
        channel.basicqos(1);
        delivercallback delivercallback = (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), "utf-8");
            system.out.println(" [x-1] received '" + delivery.getenvelope().getroutingkey() + "':'" + message + "'");
        };
        channel.basicconsume(queue_name, true, delivercallback, consumertag -> {
        });
    }
}

package topic;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;
import com.rabbitmq.client.delivercallback;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;

public class receiver2 {

    private final static string queue_name = "queue_topic2";
    private final static string exchange_name = "exchange_topic";
    private final static string exchange_type = "topic";

    public static void main(string[] args) throws ioexception, interruptedexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.queuedeclare(queue_name, false, false, false, null);
        // *号代表单个单词，可以接收key.1
        channel.queuebind(queue_name, exchange_name, "*.*");
        // #号代表多个单词，可以接收key.1.2
        channel.queuebind(queue_name, exchange_name, "*.#");
        channel.basicqos(1);
        delivercallback delivercallback = (consumertag, delivery) -> {
            string message = new string(delivery.getbody(), "utf-8");
            system.out.println(" [x-2] received '" + delivery.getenvelope().getroutingkey() + "':'" + message + "'");
        };
        channel.basicconsume(queue_name, true, delivercallback, consumertag -> {
        });
    }
}

package topic;

import com.rabbitmq.client.channel;
import com.rabbitmq.client.connection;
import com.rabbitmq.client.connectionfactory;

import java.io.ioexception;
import java.util.concurrent.timeoutexception;


public class sender {

    private final static string exchange_name = "exchange_topic";
    private final static string exchange_type = "topic";

    public static void main(string[] args) throws ioexception, timeoutexception {
        connectionfactory factory = new connectionfactory();
        factory.sethost("localhost");
        factory.setport(5672);
        connection connection = factory.newconnection();
        channel channel = connection.createchannel();
        channel.exchangedeclare(exchange_name, exchange_type);
        string message = "topics model message with key.1";
        channel.basicpublish(exchange_name, "key.1", null, message.getbytes());
        system.out.println("[x] sent '" + message + "'");
        string message2 = "topics model message with key.1.2";
        channel.basicpublish(exchange_name, "key.1.2", null, message2.getbytes());
        system.out.println("[x] sent '" + message2 + "'");
        channel.close();
        connection.close();
    }
}

[x] sent 'topics model message with key.1'
[x] sent 'topics model message with key.1.2'

[x-1] received 'key.1':'topics model message with key.1'

[x-2] received 'key.1':'topics model message with key.1'
[x-2] received 'key.1.2':'topics model message with key.1.2'

3、四种交换机介绍

• 直连交换机( direct exchange )：具有路由功能的交换机，绑定到此交换机的时候需要指定一个

  routing_key，交换机发送消息的时候需要 routing_key，会将消息发送道对应的队列。

• 扇形交换机( fanout exchange )：广播消息到所有队列，没有任何处理，速度最快。

• 主题交换机( topic exchange )：在直连交换机基础上增加模式匹配，也就是对 routing_key 进行模式匹配，

  * 代表一个单词，# 代表多个单词。

• 首部交换机( headers exchange )：忽略 routing_key，使用 headers 信息(一个 hash 的数据结构)进行匹

  配，优势在于可以有更多更灵活的匹配规则。