当前位置：首页 » 《随便一记》 » 正文

RabbitMQ 工作模式使用案例之（发布订阅模式、路由模式、通配符模式）

23 人参与  2024年12月18日 10:01  分类 : 《随便一记》  评论

点击全文阅读


Hi~！这里是奋斗的明志，很荣幸您能阅读我的文章，诚请评论指点，欢迎欢迎 ~~
??个人主页：奋斗的明志
??所属专栏：RabbitMQ

?本系列文章为个人学习笔记，在这里撰写成文一为巩固知识，二为展示我的学习过程及理解。文笔、排版拙劣，望见谅。

在这里插入图片描述

发布订阅模式、路由模式、通配符模式

一、Publish/Subscribe(发布/订阅)1、引入依赖2、编写配置类3、编写生产者代码4、编写消费者代码 二、Routing (路由模式)1、引入依赖2、编写配置类3、编写生产者代码4、编写消费者代码 三、Topics (通配符模式)1、引入依赖2、编写配置类3、编写生产者代码4、编写消费者代码

一、Publish/Subscribe(发布/订阅)

在发布/订阅模型中，多了一个Exchange角色.
Exchange 常见有三种类型, 分别代表不同的路由规则
a) Fanout:广播，将消息交给所有绑定到交换机的队列 (Publish/Subscribe模式)
b) Direct:定向，把消息交给符合指定routing key的队列(Routing模式)
c) Topic:通配符，把消息交给符合routing pattern(路由模式)的队列(Topics模式)
也就分别对应不同的工作模式

在这里插入图片描述

1、引入依赖

<dependencies>        <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->        <dependency>            <groupId>com.rabbitmq</groupId>            <artifactId>amqp-client</artifactId>            <version>5.20.0</version>        </dependency></dependencies>

2、编写配置类

package rabbitmq.constant;public class Constants {    public static final String HOST = "123.57.16.61";    public static final Integer PORT = 5672;    public static final String USERNAME = "study";    public static final String PASSWORD = "study";    public static final String VIRTUAL_HOST = "bite";    //发布订阅模式    public static final String FANOUT_EXCHANGE = "fanout.exchange";    public static final String FANOUT_QUEUE1 = "fanout.queue1";    public static final String FANOUT_QUEUE2 = "fanout.queue2";}

3、编写生产者代码

package rabbitmq.fanout;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明交换机        /**         * 交换机名称，交换机类型，开启可持久化（关机数据不会丢失）         */        channel.exchangeDeclare(Constants.FANOUT_EXCHANGE, BuiltinExchangeType.FANOUT, true);        //声明队列        //queueDeclare 队列声明        channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);        channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);        //交换机和队列进行绑定        channel.queueBind(Constants.FANOUT_QUEUE1, Constants.FANOUT_EXCHANGE, "");        channel.queueBind(Constants.FANOUT_QUEUE2, Constants.FANOUT_EXCHANGE, "");        //发布消息        String msg = "hello fanout...";        //basicPublish (基础发布)        channel.basicPublish(Constants.FANOUT_EXCHANGE, "", null, msg.getBytes());        System.out.println("消息发送成功！！！");        //关闭资源        channel.close();        connection.close();    }}

点击运行：

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

4、编写消费者代码

消费者1：

package rabbitmq.fanout;import com.rabbitmq.client.*;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer1 {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明队列        //queueDeclare 队列声明 (也可以省略)        channel.queueDeclare(Constants.FANOUT_QUEUE1, true, false, false, null);        //消费消息        DefaultConsumer consumer = new DefaultConsumer(channel){            @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("接收到消息：" + new String(body));            }        };        channel.basicConsume(Constants.FANOUT_QUEUE1,true,consumer);    }}

消费者2：

package rabbitmq.fanout;import com.rabbitmq.client.*;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明队列        //queueDeclare 队列声明 (也可以省略)        channel.queueDeclare(Constants.FANOUT_QUEUE2, true, false, false, null);        //消费消息        DefaultConsumer consumer = new DefaultConsumer(channel){            @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("接收到消息：" + new String(body));            }        };        channel.basicConsume(Constants.FANOUT_QUEUE2,true,consumer);    }}

在这里插入图片描述

二、Routing (路由模式)

队列和交换机的绑定, 不能是任意的绑定了, 而是要指定⼀个BindingKey(RoutingKey的⼀种)
消息的发送方在向 Exchange 发送消息时, 也需要指定消息的 RoutingKey
Exchange也不再把消息交给每⼀个绑定的key, 而是根据消息的RoutingKey进行判断, 只有队列绑定时的BindingKey和发送消息的RoutingKey 完全⼀致, 才会接收到消息

在这里插入图片描述

1、引入依赖

<dependencies>        <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->        <dependency>            <groupId>com.rabbitmq</groupId>            <artifactId>amqp-client</artifactId>            <version>5.20.0</version>        </dependency></dependencies>

2、编写配置类

package rabbitmq.constant;public class Constants {    public static final String HOST = "123.57.16.61";    public static final Integer PORT = 5672;    public static final String USERNAME = "study";    public static final String PASSWORD = "study";    public static final String VIRTUAL_HOST = "bite";    //路由模式    public static final String DIRECT_EXCHANGE = "direct.exchange";    public static final String DIRECT_QUEUE1 = "direct.queue1";    public static final String DIRECT_QUEUE2 = "direct.queue2";}

3、编写生产者代码

package rabbitmq.direct;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明交换机        /**         * 交换机名称，交换机类型，开启可持久化（关机数据不会丢失）         */        channel.exchangeDeclare(Constants.DIRECT_EXCHANGE, BuiltinExchangeType.DIRECT, true);        //声明队列        //queueDeclare 队列声明        channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);        channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);        //交换机和队列进行绑定        channel.queueBind(Constants.DIRECT_QUEUE1, Constants.DIRECT_EXCHANGE, "a");        channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "a");        channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "b");        channel.queueBind(Constants.DIRECT_QUEUE2, Constants.DIRECT_EXCHANGE, "c");        //发布消息        String msg_a = "hello direct, my routingkey is a...";        //basicPublish (基础发布)        channel.basicPublish(Constants.DIRECT_EXCHANGE, "", null, msg_a.getBytes());        String msg_b = "hello direct, my routingkey is b...";        channel.basicPublish(Constants.DIRECT_EXCHANGE, "", null, msg_b.getBytes());        String msg_c = "hello direct, my routingkey is c...";        channel.basicPublish(Constants.DIRECT_EXCHANGE, "", null, msg_c.getBytes());        System.out.println("消息发送成功！！！");        //关闭资源        channel.close();        connection.close();    }}

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

4、编写消费者代码

消费者1：

package rabbitmq.direct;import com.rabbitmq.client.*;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer1 {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明队列        //queueDeclare 队列声明 (也可以省略)        channel.queueDeclare(Constants.DIRECT_QUEUE1, true, false, false, null);        //消费消息        DefaultConsumer consumer = new DefaultConsumer(channel){            @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("接收到消息：" + new String(body));            }        };        channel.basicConsume(Constants.DIRECT_QUEUE1,true,consumer);    }}

在这里插入图片描述

消费者2：

package rabbitmq.direct;import com.rabbitmq.client.*;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明队列        //queueDeclare 队列声明 (也可以省略)        channel.queueDeclare(Constants.DIRECT_QUEUE2, true, false, false, null);        //消费消息        DefaultConsumer consumer = new DefaultConsumer(channel){            @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("接收到消息：" + new String(body));            }        };        channel.basicConsume(Constants.DIRECT_QUEUE2,true,consumer);    }}

在这里插入图片描述

三、Topics (通配符模式)

Topics 和Routing模式的区别是:

topics 模式使用的交换机类型为topic(Routing模式使用的交换机类型为direct)topic 类型的交换机在匹配规则上进行了扩展, Binding Key⽀持通配符匹配(direct类型的交换机路由规则是BindingKey和RoutingKey完全匹配)

在这里插入图片描述

在topic类型的交换机在匹配规则上, 有些要求:

RoutingKey 是⼀系列由点( . )分隔的单词, ⽐如 " stock.usd.nyse ", " nyse.vmw ",
" quick.orange.rabbit "BindingKey 和RoutingKey⼀样, 也是点( . )分割的字符串.Binding Key中可以存在两种特殊字符串, 用于模糊匹配 
* 表⽰⼀个单词 # 表⽰多个单词(0-N个)

比如:
• Binding Key 为"d.a.b" 会同时路由到Q1 和Q2
• Binding Key 为"d.a.f" 会路由到Q1
• Binding Key 为"c.e.f" 会路由到Q2
• Binding Key 为"d.b.f" 会被丢弃, 或者返回给⽣产者(需要设置mandatory参数)

1、引入依赖

<dependencies>        <!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client -->        <dependency>            <groupId>com.rabbitmq</groupId>            <artifactId>amqp-client</artifactId>            <version>5.20.0</version>        </dependency></dependencies>

2、编写配置类

package rabbitmq.constant;public class Constants {    public static final String HOST = "123.57.16.61";    public static final Integer PORT = 5672;    public static final String USERNAME = "study";    public static final String PASSWORD = "study";    public static final String VIRTUAL_HOST = "bite";    //通配符模式    public static final String TOPIC_EXCHANGE = "topic.exchange";    public static final String TOPIC_QUEUE1 = "topic.queue1";    public static final String TOPIC_QUEUE2 = "topic.queue2";

3、编写生产者代码

package rabbitmq.topic;import com.rabbitmq.client.BuiltinExchangeType;import com.rabbitmq.client.Channel;import com.rabbitmq.client.Connection;import com.rabbitmq.client.ConnectionFactory;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Producer {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明交换机        /**         * 交换机名称，交换机类型，开启可持久化（关机数据不会丢失）         */        channel.exchangeDeclare(Constants.TOPIC_EXCHANGE, BuiltinExchangeType.TOPIC, true);        //声明队列        //queueDeclare 队列声明        channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);        channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);        //交换机和队列进行绑定        channel.queueBind(Constants.TOPIC_QUEUE1, Constants.TOPIC_EXCHANGE, "*.a.*");        channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "*.*.b");        channel.queueBind(Constants.TOPIC_QUEUE2, Constants.TOPIC_EXCHANGE, "c.#");        //发布消息        String msg_a = "hello topic, my routingkey is ae.a.f...";        //basicPublish (基础发布)        channel.basicPublish(Constants.TOPIC_EXCHANGE, "ae.a.f", null, msg_a.getBytes());        String msg_b = "hello topic, my routingkey is ef.a.b...";        channel.basicPublish(Constants.TOPIC_EXCHANGE, "ef.a.b", null, msg_b.getBytes());        String msg_c = "hello topic, my routingkey is c.ef.b...";        channel.basicPublish(Constants.TOPIC_EXCHANGE, "c.ef.b", null, msg_c.getBytes());        System.out.println("消息发送成功！！！");        //关闭资源        channel.close();        connection.close();    }}

4、编写消费者代码

消费者1：

package rabbitmq.topic;import com.rabbitmq.client.*;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer1 {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明队列        //queueDeclare 队列声明 (也可以省略)        channel.queueDeclare(Constants.TOPIC_QUEUE1, true, false, false, null);        //消费消息        DefaultConsumer consumer = new DefaultConsumer(channel){            @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("接收到消息：" + new String(body));            }        };        channel.basicConsume(Constants.TOPIC_QUEUE1,true,consumer);    }}

在这里插入图片描述

消费者2：

package rabbitmq.topic;import com.rabbitmq.client.*;import rabbitmq.constant.Constants;import java.io.IOException;import java.util.concurrent.TimeoutException;public class Consumer2 {    public static void main(String[] args) throws IOException, TimeoutException {        ConnectionFactory connectionFactory = new ConnectionFactory();        connectionFactory.setHost(Constants.HOST);        connectionFactory.setPort(Constants.PORT);        connectionFactory.setUsername(Constants.USERNAME);        connectionFactory.setPassword(Constants.PASSWORD);        connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST);        Connection connection = connectionFactory.newConnection();        Channel channel = connection.createChannel();        //声明队列        //queueDeclare 队列声明 (也可以省略)        channel.queueDeclare(Constants.TOPIC_QUEUE2, true, false, false, null);        //消费消息        DefaultConsumer consumer = new DefaultConsumer(channel){            @Override            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {                System.out.println("接收到消息：" + new String(body));            }        };        channel.basicConsume(Constants.TOPIC_QUEUE2,true,consumer);    }}

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述


点击全文阅读


本文链接：http://m.zhangshiyu.com/post/202899.html

<< 上一篇 下一篇 >>

  • 评论（0）
  • 赞助本站

◎欢迎参与讨论，请在这里发表您的看法、交流您的观点。

最新文章

  • 心机女勾勾手指，大佬们死去活来完整版(黎芥庄可迁)抖音热文_《心机女勾勾手指，大佬们死去活来完整版》最新章节免费在线阅读 -
  • 《不装了，我就是仙二代》全集免费在线阅读（叶云龙周暮雪）
  • 主角叫叶休叶小说被猜忌后，他重掌天下全文免费阅读
  • 成了大明长孙后，他被迫继承皇位免费阅读，成了大明长孙后，他被迫继承皇位朱烨蓝玉
  • 被抛弃后，她成了豪门真千金免费阅读，被抛弃后，她成了豪门真千金章节目录
  • 都市：透视邪少叶辰李霸天小说在线全文免费阅读
  • 完整版《阴纹师》全章节阅读
  • 道求无上最新章节，道求无上全文在线阅读
  • 道心种魔（秦少风大罗金仙）在线免费阅读
  • 我，成魔后，一剑撼天全集(莫湘王腾)小说免费阅读完整版_(我，成魔后，一剑撼天全集)最新章节免费在线阅读 -
  • 拜托，有两个大佬崽崽超酷的！小说，拜托，有两个大佬崽崽超酷的！在线阅读
  • 陪你撒野小说免费资源，陪你撒野在线阅读

    关于我们 | 我要投稿 | 免责申明

    Copyright © 2020-2022 ZhangShiYu.com Rights Reserved.豫ICP备2022013469号-1