03-RabbitMQ的工作模式

lannntuuu 2023-03-28 原文

RabbitMQ的工作模式

一、模式概述

RabbitMQ提供了6种工作模式：简单模式、工作队列模式、订阅模式、路由模式、通配符模式、远程调用模式

其中远程调用模式（RPC）暂不作介绍。

官网对于模式介绍：https://www.rabbitmq.com/getstarted.html

二、简单模式 HelloWorld

2.1、模式说明

在上图的模型中，有以下概念：

P：生产者，也就是要发送消息的程序
C：消费者，消息的接收者，会一直等待消息到来
Queue：消息队列，图中红色部分。类似一个邮箱，可以缓存消息；生产者向其中投递消息，消费者从其中取出消息。

2.2、代码举例

需求：使用简单模式完成消息传递

步骤：

创建工程（生产者、消费者）
分别添加依赖
编写生产者发送消息
编写消费者接收消息

1、创建工程

创建空项目RabbitMQ，然后在RabbitMQ下面创建生产者rabbitmq-producer、消费者rabbitmq-consumer模块（New Module - Build System选Maven）

2、添加依赖和编译插件

这里生产者和消费者是一样的

maven-compiler-plugin 插件是一个 Maven 插件，用来编译项目代码。

自从3.0开始默认的编译器是 javax.tools.JavaCompiler，用来编译 Java 源码；

如果你想强制插件使用 javac 编译器，你必须配置插件的属性 forceJavacCompilerUse；

还要注意，当前默认源（source）设置为 1.8，默认目标(target)设置为 1.8。独立运行 Maven 和 JDK，可以通过 source 和 target 选项更改他们的默认值；

插件设置的各种参数信息请查看

http://maven.apache.org/plugins/maven-compiler-plugin/compile-mojo.html#forceJavacCompilerUse

<dependencies>
    <!--rabbitmq java 客户端-->
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.6.0</version>
    </dependency>
</dependencies>


<build>
    <plugins>
        <plugin>
            <groupId>org.apache.maven.plugins</groupId>
            <artifactId>maven-compiler-plugin</artifactId>
            <version>3.8.1</version>
            <configuration>
                <source>1.8</source>
                <target>1.8</target>
            </configuration>
        </plugin>
    </plugins>
</build>

3、编写生产者发送消息

public class Producer_HelloWorld {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建队列 Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1、queue：队列名称
            2、durable：是否持久化，当mq重启之后，还在
            3、exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当Connection关闭时，是否删除队列
            4、autoDelete：是否自动化删除。当没有Consumer时，自动删掉
            5、arguments：参数。
         */
        
        // 如果没有一个名字叫 Hello_World 的队列，则会创建该队列，如果有则不会创建
        channel.queueDeclare("Hello_World", true, false, false, null);
        
        // 6、发送消息
        /*
        basicPublish(String exchange, String routingKey, AMQP.BasicProperties props, byte[] body)
        参数：
            1、exchange：交换机名称。简单模式下交换机会使用默认的 ""
            2、routingKey：路由名称
            3、props：配置信息
            4、body：发送消息数据
         */
        String body = "Hello RabbitMQ~~~";
        
        channel.basicPublish("", "Hello_World", null, body.getBytes());

        // 7、释放资源
        channel.close();
        connection.close();
    }
}

4、编写消费者接收消息

public class Consumer_HelloWorld {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建队列 Queue
        /*
        queueDeclare(String queue, boolean durable, boolean exclusive, boolean autoDelete, Map<String, Object> arguments)
        参数：
            1、queue：队列名称
            2、durable：是否持久化，当mq重启之后，还在
            3、exclusive：
                * 是否独占。只能有一个消费者监听这队列
                * 当Connection关闭时，是否删除队列
            4、autoDelete：是否自动化删除。当没有Consumer时，自动删掉
            5、arguments：参数。
         */
        
        // 如果没有一个名字叫 Hello_World 的队列，则会创建该队列，如果有则不会创建
//        channel.queueDeclare("Hello_World", true, false, false, null);
        
        // 6、接收消息
        /*
        basicConsume(String queue, boolean autoAck, Consumer callback)
        参数：
            1、queue：队列名称
            2、autoAck：是否自动确认
            2、callback：回调对象
         */

        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /*
                回调方法，当收到消息之后，会自动执行该方法
                
                1、consumerTag：标识
                2、envelope：获取一些信息，交换机，路由key...
                3、properties：配置信息
                4、body：数据
             */

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("consumerTag：" + consumerTag);
                System.out.println("Exchange：" + envelope.getExchange());
                System.out.println("RoutingKey：" + envelope.getRoutingKey());
                System.out.println("properties：" + properties);
                System.out.println("body：" + new String(body));
            }
        };
        
        channel.basicConsume("Hello_World", true, consumer);

        // 7、释放资源 不需要
    }
}

5、结果

启动生产者，在虚拟机中创建队列，像虚拟机中指定队列发送消息
启动消费者，与虚拟机建立连接，监听指定队列，接收消息

三、工作队列模式（Work Queues）

3.1、模式说明

Work Queues：与简单模式相比，多了一个或多个消费者，多个消费者共同消费同一个队列中的消息。
应用场景：对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。

3.2、代码举例

Work Queues与简单模式的代码几乎是一样的。可以完全复制，并多复制一个消费者进行多个消费者同时对消费者的测试。

1、生产者

public class Producer_WorkQueues {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建队列 Queue
        channel.queueDeclare("Work_Queues", true, false, false, null);
        
        // 6、发送消息
        for (int i = 0; i < 10; i++) {
            String body = i + "Hello RabbitMQ~~~";

            channel.basicPublish("", "Work_Queues", null, body.getBytes());
        }

        // 7、释放资源
        channel.close();
        connection.close();
    }
}

2、消费者1/消费者2

public class Consumer_WorkQueues1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();
        
        // 5、创建队列
        channel.queueDeclare("Work_Queues", true, false, false, null);
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            /*
                回调方法，当收到消息之后，会自动执行该方法
                
                1、consumerTag：标识
                2、envelope：获取一些信息，交换机，路由key...
                3、properties：配置信息
                4、body：数据
             */

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
            }
        };
        
        channel.basicConsume("Work_Queues", true, consumer);

        // 7、释放资源 不需要
    }
}

3、结果

先启动生产者再启动消费者会导致第一个启动的消费者把消息全消费了。

所以先启动两个消费者，再启动生产者，这样就可以看到两个消费者是轮询着消费。

当然也可以把发送消息单独写一个类出来，或者延时执行。

4、小结

在一个队列中如果有多个消费者，那么对于同一个消息来说消费者之间是竞争关系。
Work Queues对于任务过重或任务较多情况使用工作队列可以提高任务处理的速度。例如：短信服务部署多个，只需要有一个节点成功发送即可。

四、订阅模式（Publish/Subscribe）

4.1、模式说明

在订阅模型中，多了一个Exchange角色，而且过程略有变化：

P：生产者，也就是要发送消息的程序，但是不再直接发送到队列中，而是发给X（交换机）
C：消费者，消息的接收者，会一直等待消息到来
Queue：消息队列，接收消息、缓存消息
X：Exchange（交换机），一方面，接收生产者发送的消息。另一方面，直到如何处理消息，例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作，取决于Exchange的类型。

Exchange有常见以下3种类型：

Fanout：广播，将消息交给所有绑定到交换机的队列

Direct：定向，把消息交给符合Routing Key的队列

Topic：通配符，把消息交给符合Routing Pattern（路由模式）的队列

Exchange（交换机）只负责转发消息，不具备存储消息的能力，因此如果没有任何队列与Exchange绑定，或者没有符合路由规则的队列，那么消息会丢失！

4.2、代码举例

1、生产者

public class Producer_PubSub {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建交换机
        /*
        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
        参数：
            1、exchange：交换机名称
            2、type：交换机类型    
                DIRECT("direct")：定向
                FANOUT("fanout")：扇形（广播），发生消息到每一个与之绑定队列。
                TOPIC("topic")：通配符的方式
                HEADERS("headers")：参数匹配
            3、durable：是否持久化
            4、autoDelete：自动删除
            5、internal：内部使用。一般false
            6、arguments：参数
            
         */
        String exchangeName = "test_fanout";
        
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.FANOUT, true, false, false, null);
        
        // 6、创建队列
        String queue1Name = "test_fanout_queue1";
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue1Name, true, false, false, null);
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 7、绑定队列和交换机
        /*
        queueBind(String queue, String exchange, String routingKey)
        参数：
            1、queue：队列名称
            2、exchange：交换机名称
            3、routingKey：路由键，绑定规则
                如果交换机的类型为fanout，routingkey设置为""
         */
        channel.queueBind(queue1Name, exchangeName, "");
        channel.queueBind(queue2Name, exchangeName, "");
        
        // 8、发送消息
        String body = "日志信息：张三调用了findAll方法...日志级别：info...";
        
        channel.basicPublish(exchangeName, "", null, body.getBytes());
        // 9、释放资源
        channel.close();
        connection.close();
    }
}

2、消费者

消费者1

public class Consumer_PubSub1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建队列 Queue
        String queue1Name = "test_fanout_queue1";
        channel.queueDeclare(queue1Name, true, false, false, null);
        
        // 6、接收消息

        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        
        channel.basicConsume(queue1Name, true, consumer);

        // 7、释放资源 不需要
    }
}

消费者2

public class Consumer_PubSub2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建队列 Queue
        String queue2Name = "test_fanout_queue2";
        channel.queueDeclare(queue2Name, true, false, false, null);

        // 6、接收消息

        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("将日志信息保存到数据库.....");
            }
        };
        
        channel.basicConsume(queue2Name, true, consumer);

        // 7、释放资源 不需要
    }
}

3、结果

先启动生产者，会创建广播模式的交换机和队列，然后绑定交换机和队列。之后发送消息，通过交换机广播发送到两个绑定的队列中。

两个消费者收到消息然后分别执行自己的后续逻辑。

4、小结

交换机需要与队列进行绑定，绑定之后，一个消息可以被多个消费者收到。
发布订阅模式与工作队列模式的区别：
- 工作队列模式不要定义交换机，而发布/订阅模式需要定义交换机
- 工作队列模式的生产方是面向队列发送消息（底层使用默认交换机），发布/订阅模式的生产方是面向交换机发送消息。
- 发布/订阅模式需要设置队列和交换机的绑定，工作队列模式不需要设置，实际上工作队列模式会将队列绑定到默认的交换机。

五、路由模式（Routing）

5.1、模式说明

队列与交换机的绑定，不能是任意绑定了，而是要指定一个RoutingKey（路由键）。
消息的发送方在向Exchange发送消息时，也必须指定消息的RoutingKey。
Exchange不再把消息交给每一个绑定的队列，而是根据消息的RoutingKey来进行判断，只有队列的RoutingKey与消息的RoutingKey完全一致，才会接受到消息。

图解：

P：生产者，向Exchange发送消息，发送消息时，会指定一个RoutingKey
X：交换机，接收生产者的消息，然后把消息递交给与RoutingKey完全匹配的队列
C1：消费者，其所在队列指定了需要RoutingKey为error的消息
C2：消费者，其所在队列指定了需要RoutingKey为info、warning、error的消息

5.2、代码举例

1、生产者

public class Producer_Routing {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建交换机
        String exchangeName = "test_direct";
        
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT, true, false, false, null);
        
        // 6、创建队列
        String queue1Name = "test_direct_queue1";
        String queue2Name = "test_direct_queue2";
        channel.queueDeclare(queue1Name, true, false, false, null);
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 7、绑定队列和交换机
        /*
        queueBind(String queue, String exchange, String routingKey)
        参数：
            1、queue：队列名称
            2、exchange：交换机名称
            3、routingKey：路由键，绑定规则
                如果交换机的类型为fanout，routingkey设置为""
         */
        // 队列1绑定 info warning error
        channel.queueBind(queue1Name, exchangeName, "info");
        channel.queueBind(queue1Name, exchangeName, "warning");
        channel.queueBind(queue1Name, exchangeName, "error");
        // 队列2绑定 error
        channel.queueBind(queue2Name, exchangeName, "error");
        
        // 8、发送消息
        String body = "日志信息：张三调用了delete方法...出错误了...日志级别：error...";
        
        channel.basicPublish(exchangeName, "error", null, body.getBytes());

        body = "日志信息：张三调用了findAll方法...日志级别：info...";

        channel.basicPublish(exchangeName, "info", null, body.getBytes());
        // 9、释放资源
        channel.close();
        connection.close();
    }
}

2、消费者

消费者1

public class Consumer_Routing1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建队列 Queue
        String queue1Name = "test_direct_queue1";
        channel.queueDeclare(queue1Name, true, false, false, null);
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        
        channel.basicConsume(queue1Name, true, consumer);

        // 7、释放资源 不需要
    }
}

消费者2

public class Consumer_Routing2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建队列 Queue
        String queue2Name = "test_direct_queue2";
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("将日志信息保存到数据库.....");
            }
        };
        
        channel.basicConsume(queue2Name, true, consumer);

        // 7、释放资源 不需要
    }
}

3、结果

先启动生产者，会创建路由模式的交换机和队列，然后绑定交换机和队列并指定对应的路由键。之后发送消息，通过交换机判断消息带的路由键发送到两个绑定的队列中。

两个消费者收到消息然后分别执行自己的后续逻辑。

4、小结

Routing模式要求队列在绑定交换机时要指定RoutingKey，消息会转发到符合RoutingKey的队列。

六、通配符模式（Topic）

6.1、模式说明

Topic类型与Direct相比，都是可以根据RoutingKey把消息路由到不同的队列。只不过Topic类型Exchange可以让队列在绑定RoutingKey的时候使用通配符！
RoutingKey一般都是由一个或多个单词组成，多个单词之间以“.”分割，例如：item.insert。
通配符规则：#匹配一个或多个词，*匹配不多不少恰好一个词，例如：item.#能够匹配item.insert.abc或者item.insert，item.*只能匹配item.insert。

右边图解：

红色Queue：绑定的是usa.#，因此凡是以usa.开头的RoutingKey都会被匹配到
黄色Queue：绑定的是#.usa，因此凡是以.news结尾的RoutingKey都会被匹配到

6.2、代码举例

1、生产者

public class Producer_Topic {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        // 5、创建交换机
        /*
        exchangeDeclare(String exchange, BuiltinExchangeType type, boolean durable, boolean autoDelete, boolean internal, Map<String, Object> arguments)
        参数：
            1、exchange：交换机名称
            2、type：交换机类型    
                DIRECT("direct")：定向
                FANOUT("fanout")：扇形（广播），发生消息到每一个与之绑定队列。
                TOPIC("topic")：通配符的方式
                HEADERS("headers")：参数匹配
            3、durable：是否持久化
            4、autoDelete：自动删除
            5、internal：内部使用。一般false
            6、arguments：参数
            
         */
        String exchangeName = "test_topic";
        
        channel.exchangeDeclare(exchangeName, BuiltinExchangeType.TOPIC, true, false, false, null);
        
        // 6、创建队列
        String queue1Name = "test_topic_queue1";
        String queue2Name = "test_topic_queue2";
        channel.queueDeclare(queue1Name, true, false, false, null);
        channel.queueDeclare(queue2Name, true, false, false, null);
        
        // 7、绑定队列和交换机
        /*
        queueBind(String queue, String exchange, String routingKey)
        参数：
            1、queue：队列名称
            2、exchange：交换机名称
            3、routingKey：路由键，绑定规则
                如果交换机的类型为fanout，routingkey设置为""
         */
        // routing key 系统的名称.日志的级别
        // 需求：所有error级别的日志存入数据库，所有order系统的日志存入数据库
        channel.queueBind(queue1Name, exchangeName, "*.*");
        channel.queueBind(queue2Name, exchangeName, "#.error");
        channel.queueBind(queue2Name, exchangeName, "order.*");
        
        // 8、发送消息
        String body = "日志信息：张三调用了delete方法...出错误了...日志级别：error...";
        
        channel.basicPublish(exchangeName, "good.address.error", null, body.getBytes());

        body = "日志信息：张三调用了findAll方法...日志级别：info...";

        channel.basicPublish(exchangeName, "good.info", null, body.getBytes());

        body = "日志信息：张三调用了update方法...日志级别：info...";

        channel.basicPublish(exchangeName, "order.info", null, body.getBytes());
        // 9、释放资源
        channel.close();
        connection.close();
    }
}

2、消费者

消费者1

public class Consumer_Topic1 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        String queue1Name = "test_topic_queue1";
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("将日志信息打印到控制台.....");
            }
        };
        
        channel.basicConsume(queue1Name, true, consumer);

        // 7、释放资源 不需要
    }
}

消费者2

public class Consumer_Topic2 {

    public static void main(String[] args) throws IOException, TimeoutException {
        
        // 1、创建连接工厂
        ConnectionFactory factory = new ConnectionFactory();

        // 2、设置参数
        factory.setHost("192.168.3.100");  // ip 默认值 localhost
        factory.setPort(5673); // 端口 默认值 5672
        factory.setVirtualHost("/"); // 虚拟机 默认值 /
        factory.setUsername("guest"); // 用户名 默认值 guest
        factory.setPassword("guest"); // 密码 默认值 guest

        // 3、创建连接 Connection
        Connection connection = factory.newConnection();

        // 4、创建频道 Channel
        Channel channel = connection.createChannel();

        String queue2Name = "test_topic_queue2";
        
        // 6、接收消息
        DefaultConsumer consumer = new DefaultConsumer(channel) {
            
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("body：" + new String(body));
                System.out.println("将日志信息保存到数据库.....");
            }
        };
        
        channel.basicConsume(queue2Name, true, consumer);

        // 7、释放资源 不需要
    }
}

3、结果

先启动生产者，会创建通配符模式的交换机和队列，然后绑定交换机和队列并指定对应的路由键。之后发送消息，通过交换机判断消息带的路由键发送到两个绑定的队列中。

两个消费者收到消息然后分别执行自己的后续逻辑。

4、小结

Topic主题模式可以实现Pub/Sub发布与订阅模式和Routing路由模式的功能，只是Topic在配置RoutingKey

的时候可以使用通配符，显得更加灵活。

七、工作模式总结

简单模式 HelloWorld

一个生产者、一个消费者，不需要设置交换机（使用默认的交换机）。
工作模式 Work Queues

一个生产者、多个消费者（竞争关系），不需要设置交换机（使用默认的交换机）。
发布订阅模式 Publish/Subscribe

需要设置类型为fanout的交换机，并且交换机和队列进行绑定，当发送消息到交换机后，交换机会将消息发送到绑定的队列。
路由模式 Routing

需要设置类型为direct的交换机，交换机和队列进行绑定，并且指定RoutingKey，当发送消息到交换机后，交换机会根据RoutingKey将消息发送到对应的队列。
通配符 Topic

需要设置类型为topic的交换机，交换机和队列进行绑定，并且指定通配符方式的RoutingKey，当发送消息到交换机后，交换机会根据RoutingKey将消息发送到对应的队列。

参考

黑马程序员RabbitMQ全套教程：https://www.bilibili.com/video/BV15k4y1k7Ep

RabbitMQ 03 strong factory channel Java

有关03-RabbitMQ的工作模式的更多相关文章

ruby-on-rails - 由于 "wkhtmltopdf"，PDFKIT 显然无法正常工作 - 2
我在从html页面生成PDF时遇到问题。我正在使用PDFkit。在安装它的过程中，我注意到我需要wkhtmltopdf。所以我也安装了它。我做了PDFkit的文档所说的一切......现在我在尝试加载PDF时遇到了这个错误。这里是错误:commandfailed:"/usr/local/bin/wkhtmltopdf""--margin-right""0.75in""--page-size""Letter""--margin-top""0.75in""--margin-bottom""0.75in""--encoding""UTF-8""--margin-left""0.75in""-
ruby-on-rails - Rails - 子类化模型的设计模式是什么？ - 2
我有一个模型:classItem项目有一个属性“商店”基于存储的值，我希望Item对象对特定方法具有不同的行为。Rails中是否有针对此的通用设计模式？如果方法中没有大的if-else语句，这是如何干净利落地完成的？最佳答案通常通过Single-TableInheritance. 关于ruby-on-rails-Rails-子类化模型的设计模式是什么？，我们在StackOverflow上找到一个类似的问题： https://stackoverflow.co
ruby-on-rails - 'compass watch' 是如何工作的/它是如何与 rails 一起使用的 - 2
我在我的项目目录中完成了compasscreate.和compassinitrails。几个问题:我已将我的.sass文件放在public/stylesheets中。这是放置它们的正确位置吗？当我运行compasswatch时，它不会自动编译这些.sass文件。我必须手动指定文件:compasswatchpublic/stylesheets/myfile.sass等。如何让它自动运行？文件ie.css、print.css和screen.css已放在stylesheets/compiled。如何在编译后不让它们重新出现的情况下删除它们？我自己编译的.sass文件编译成compiled/t
ruby - 解析 RDFa、微数据等的最佳方式是什么，使用统一的模式/词汇(例如 schema.org)存储和显示信息 - 2
我主要使用Ruby来执行此操作，但到目前为止我的攻击计划如下:使用gemsrdf、rdf-rdfa和rdf-microdata或mida来解析给定任何URI的数据。我认为最好映射到像schema.org这样的统一模式，例如使用这个yaml文件，它试图描述数据词汇表和opengraph到schema.org之间的转换:#SchemaXtoschema.orgconversion#data-vocabularyDV:name:namestreet-address:streetAddressregion:addressRegionlocality:addressLocalityphoto:i
ruby - 如何在续集中重新加载表模式？ - 2
鉴于我有以下迁移:Sequel.migrationdoupdoalter_table:usersdoadd_column:is_admin,:default=>falseend#SequelrunsaDESCRIBEtablestatement,whenthemodelisloaded.#Atthispoint,itdoesnotknowthatusershaveais_adminflag.#Soitfails.@user=User.find(:email=>"admin@fancy-startup.example")@user.is_admin=true@user.save!ende
ruby - 无法让 RSpec 工作—— 'require' : cannot load such file - 2
我花了三天的时间用头撞墙，试图弄清楚为什么简单的“rake”不能通过我的规范文件。如果您遇到这种情况:任何文件夹路径中都不要有空格!。严重地。事实上，从现在开始，您命名的任何内容都没有空格。这是我的控制台输出:(在/Users/*****/Desktop/LearningRuby/learn_ruby)$rake/Users/*******/Desktop/LearningRuby/learn_ruby/00_hello/hello_spec.rb:116:in`require':cannotloadsuchfile--hello(LoadError) 最佳
ruby-on-rails - rspec should have_select ('cars' , :options => ['volvo' , 'saab' ] 不工作 - 2
关闭。这个问题需要detailsorclarity.它目前不接受答案。想改进这个问题吗？通过editingthispost添加细节并澄清问题.关闭8年前。Improvethisquestion在首页我有:汽车:VolvoSaabMercedesAudistatic_pages_spec.rb中的测试代码:it"shouldhavetherightselect"dovisithome_pathit{shouldhave_select('cars',:options=>['volvo','saab','mercedes','audi'])}end响应是rspec./spec/request
ruby-on-rails - s3_direct_upload 在生产服务器中不工作 - 2
在Rails4.0.2中，我使用s3_direct_upload和aws-sdkgems直接为s3存储桶上传文件。在开发环境中它工作正常，但在生产环境中它会抛出如下错误，ActionView::Template::Error(noimplicitconversionofnilintoString)在View中，create_cv_url,:id=>"s3_uploader",:key=>"cv_uploads/{unique_id}/${filename}",:key_starts_with=>"cv_uploads/",:callback_param=>"cv[direct_uplo
ruby - 是否有用于序列化和反序列化各种格式的对象层次结构的模式？ - 2
给定一个复杂的对象层次结构，幸运的是它不包含循环引用，我如何实现支持各种格式的序列化？我不是来讨论实际实现的。相反，我正在寻找可能会派上用场的设计模式提示。更准确地说:我正在使用Ruby，我想解析XML和JSON数据以构建复杂的对象层次结构。此外，应该可以将该层次结构序列化为JSON、XML和可能的HTML。我可以为此使用Builder模式吗？在任何提到的情况下，我都有某种结构化数据-无论是在内存中还是文本中-我想用它来构建其他东西。我认为将序列化逻辑与实际业务逻辑分开会很好，这样我以后就可以轻松支持多种XML格式。最佳答案我最
ruby - JetBrains RubyMine 3.2.4 调试器不工作 - 2
使用Ruby1.9.2运行IDE提示说需要gemruby-debug-base19x并提供安装它。但是，在尝试安装它时会显示消息Failedtoinstallgems.Followinggemswerenotinstalled:C:/ProgramFiles(x86)/JetBrains/RubyMine3.2.4/rb/gems/ruby-debug-base19x-0.11.30.pre2.gem:Errorinstallingruby-debug-base19x-0.11.30.pre2.gem:The'linecache19'nativegemrequiresinstall

03-RabbitMQ的工作模式

RabbitMQ的工作模式

一、模式概述

二、简单模式 HelloWorld

2.1、模式说明

2.2、代码举例

1、创建工程

2、添加依赖和编译插件

3、编写生产者发送消息

4、编写消费者接收消息

5、结果

三、工作队列模式（Work Queues）

3.1、模式说明

3.2、代码举例

1、生产者

2、消费者1/消费者2

3、结果

4、小结

四、订阅模式 （Publish/Subscribe）

4.1、模式说明

4.2、代码举例

1、生产者

2、消费者

3、结果

4、小结

五、路由模式（Routing）

5.1、模式说明

5.2、代码举例

1、生产者

2、消费者

3、结果

4、小结

六、通配符模式（Topic）

6.1、模式说明

6.2、代码举例

1、生产者

2、消费者

3、结果

4、小结

七、工作模式总结

参考

有关03-RabbitMQ的工作模式的更多相关文章

随机推荐

四、订阅模式（Publish/Subscribe）