网上网城网站,长沙县工程建设质监站网站,如何自己做一个网页,寿宁建设局网站SpringAMQP
将来我们开发业务功能的时候#xff0c;肯定不会在控制台收发消息#xff0c;而是应该基于编程的方式。由于RabbitMQ采用了AMQP协议#xff0c;因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息#xff0c;都可以与RabbitMQ交互。并且RabbitMQ官方也…SpringAMQP
将来我们开发业务功能的时候肯定不会在控制台收发消息而是应该基于编程的方式。由于RabbitMQ采用了AMQP协议因此它具备跨语言的特性。任何语言只要遵循AMQP协议收发消息都可以与RabbitMQ交互。并且RabbitMQ官方也提供了各种不同语言的客户端。
但是RabbitMQ官方提供的Java客户端编码相对复杂一般生产环境下我们更多会结合Spring来使用。而Spring的官方刚好基于RabbitMQ提供了这样一套消息收发的模板工具SpringAMQP。并且还基于SpringBoot对其实现了自动装配使用起来非常方便。
SpringAmqp的官方地址
Spring AMQP
SpringAMQP提供了三个功能 自动声明队列、交换机及其绑定关系基于注解的监听器模式异步接收消息封装了RabbitTemplate工具用于发送消息 快速入门 别忘了在我们的项目中引入spring amqp的依赖。 dependenciesdependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency!--AMQP依赖包含RabbitMQ--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-amqp/artifactId/dependency!--单元测试--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactId/dependency
/dependencies 在之前的案例中我们都是经过交换机发送消息到队列不过有时候为了测试方便我们也可以直接向队列发送消息跳过交换机。
在入门案例中我们就演示这样的简单模型如图 也就是 publisher直接发送消息到队列消费者监听并处理队列中的消息 注意这种模式一般测试使用很少在生产中使用。 为了方便测试我们在rabbitMQ控制台创建名为 simple.queue 的队列。 添加队列后查看 接下来我们就可以利用Java代码收发消息了。 消息发送 在我们的项目application.yml 中 添加关于rabbitmq的配置信息。 spring:rabbitmq:host: 123.56.247.70 # 你的虚拟机IPport: 5672 # rabbitMQ端口virtual-host: /sde # 虚拟机名称username: sundaoen # 用户名password: 8888888888 # 密码 编写测试类 在我们项目的publisher中创建测试类并且利用 RabbitTemplate 发送消息。 SpringBootTest
public class TestSendMessage {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSimpleQueue(){// 1 队列名称String queueName simple.queue;// 2 消息String message hello simple.queue;// 3 发送消息rabbitTemplate.convertAndSend(queueName,message);}
} 打开控制台可以看到消息已经发送到队列中 看看消息内容 接下来我们再来实现消息接收。 消息接收 同样的道理也是先配置MQ地址。在application.yml 中 spring:rabbitmq:host: 123.56.247.70 # 你的虚拟机IPport: 5672 # rabbitMQ端口virtual-host: /sde # 虚拟机名称username: sundaoen # 用户名password: 8888888888 # 密码
在consumer服务中编写监听器类并利用RabbitListener实现消息的接收。
Slf4j
Component
public class SimpleQueueListener {/*利用RabbitListener注解可以监听到对应队列的消息一旦监听的队列有消息就会回调当前方法在方法中接收消息并消费处理消息*/RabbitListener(queues simple.queue)public void listenerSimpleQueue(String msg){System.out.println(SpringRabbitListener 监听到 simple.queue 队列中的消息是 msg);}
} WorkQueue模型
Work queues任务模型。简单来说就是让多个消费者绑定到一个队列共同消费队列中的消息。 当消息处理比较耗时的时候可能生产消息的速度会远远大于消息的消费速度。长此以往消息就会堆积越来越多无法及时处理。
此时就可以使用work 模型多个消费者共同处理消息处理消息处理的速度就能大大提高了。
接下来我们就来模拟这样的场景。
首先我们在控制台创建一个新的队列命名为work.queue 添加后的效果 消息发送
这次我们循环发送模拟大量消息堆积现象。
在publisher服务中的WorkQueueSendTest类中添加一个测试方法 SpringBootTest
public class WorkQueueSendTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendWorkQueue() throws InterruptedException {// 1 队列名称String queueName work.queue;// 2 消息String message hello work.queue-;// 3 发送消息for (int i 1; i 50; i) {// 每隔20毫秒发送一条消息相当于一秒发送50条消息。rabbitTemplate.convertAndSend(queueName,message i);Thread.sleep(20);}}
}
可以看到在work.queue 队列中有50条消息。 消息接收
要模拟多个消费者绑定同一个队列我们在consumer服务中的listener包中。新增WorkQueueListener类并添加2个新的方法 Slf4j
Component
public class WorkQueueListener {/*实现两个消费 work.queue的监听消费消息的方法一个方法消费后沉睡 20毫秒一个消息消费后沉睡200毫秒*/RabbitListener(queues work.queue)public void listenerWorkQueue1(String msg){System.out.println(消费者1接收到消息 msg 时间 LocalDateTime.now());try {Thread.sleep(20); // 沉睡20毫秒 1秒是1000毫秒等于1秒处理50条消息} catch (InterruptedException e) {e.printStackTrace();}}RabbitListener(queues work.queue)public void listenerWorkQueue2(String msg){System.out.println(***消费者2接收到消息 msg 时间 LocalDateTime.now());try {Thread.sleep(200); // 沉睡200毫秒 1秒是1000毫秒等于1秒处理5条消息} catch (InterruptedException e) {e.printStackTrace();}}}
注意到这两消费者都设置了Thead.sleep模拟任务耗时
消费者1 sleep了20毫秒相当于每秒钟处理50个消息消费者2 sleep了200毫秒相当于每秒处理5个消息
测试看结果
消费者1接收到消息hello work.queue-1 时间2025-02-05T14:53:00.928905400
***消费者2接收到消息hello work.queue-2 时间2025-02-05T14:53:00.947629900
消费者1接收到消息hello work.queue-3 时间2025-02-05T14:53:00.977764800
消费者1接收到消息hello work.queue-5 时间2025-02-05T14:53:01.039608
消费者1接收到消息hello work.queue-7 时间2025-02-05T14:53:01.101242200
消费者1接收到消息hello work.queue-9 时间2025-02-05T14:53:01.160396600
***消费者2接收到消息hello work.queue-4 时间2025-02-05T14:53:01.161396900
消费者1接收到消息hello work.queue-11 时间2025-02-05T14:53:01.231704200
消费者1接收到消息hello work.queue-13 时间2025-02-05T14:53:01.281879300
消费者1接收到消息hello work.queue-15 时间2025-02-05T14:53:01.347333400
***消费者2接收到消息hello work.queue-6 时间2025-02-05T14:53:01.376528100
消费者1接收到消息hello work.queue-17 时间2025-02-05T14:53:01.407569700
消费者1接收到消息hello work.queue-19 时间2025-02-05T14:53:01.464497900
消费者1接收到消息hello work.queue-21 时间2025-02-05T14:53:01.525121200
消费者1接收到消息hello work.queue-23 时间2025-02-05T14:53:01.587589500
***消费者2接收到消息hello work.queue-8 时间2025-02-05T14:53:01.589591300
消费者1接收到消息hello work.queue-25 时间2025-02-05T14:53:01.647549500
消费者1接收到消息hello work.queue-27 时间2025-02-05T14:53:01.709757900
消费者1接收到消息hello work.queue-29 时间2025-02-05T14:53:01.768879300
***消费者2接收到消息hello work.queue-10 时间2025-02-05T14:53:01.801437800
消费者1接收到消息hello work.queue-31 时间2025-02-05T14:53:01.829539900
消费者1接收到消息hello work.queue-33 时间2025-02-05T14:53:01.895907400
消费者1接收到消息hello work.queue-35 时间2025-02-05T14:53:01.950810
消费者1接收到消息hello work.queue-37 时间2025-02-05T14:53:02.011575
***消费者2接收到消息hello work.queue-12 时间2025-02-05T14:53:02.014526300
消费者1接收到消息hello work.queue-39 时间2025-02-05T14:53:02.073814400
消费者1接收到消息hello work.queue-41 时间2025-02-05T14:53:02.142812400
消费者1接收到消息hello work.queue-43 时间2025-02-05T14:53:02.199522100
***消费者2接收到消息hello work.queue-14 时间2025-02-05T14:53:02.228114600
消费者1接收到消息hello work.queue-45 时间2025-02-05T14:53:02.255591100
消费者1接收到消息hello work.queue-47 时间2025-02-05T14:53:02.315954800
消费者1接收到消息hello work.queue-49 时间2025-02-05T14:53:02.377632900
***消费者2接收到消息hello work.queue-16 时间2025-02-05T14:53:02.440855300
***消费者2接收到消息hello work.queue-18 时间2025-02-05T14:53:02.654015100
***消费者2接收到消息hello work.queue-20 时间2025-02-05T14:53:02.867783300
***消费者2接收到消息hello work.queue-22 时间2025-02-05T14:53:03.080905400
***消费者2接收到消息hello work.queue-24 时间2025-02-05T14:53:03.296731200
***消费者2接收到消息hello work.queue-26 时间2025-02-05T14:53:03.512099400
***消费者2接收到消息hello work.queue-28 时间2025-02-05T14:53:03.725353500
***消费者2接收到消息hello work.queue-30 时间2025-02-05T14:53:03.939706400
***消费者2接收到消息hello work.queue-32 时间2025-02-05T14:53:04.152588100
***消费者2接收到消息hello work.queue-34 时间2025-02-05T14:53:04.367337200
***消费者2接收到消息hello work.queue-36 时间2025-02-05T14:53:04.581549200
***消费者2接收到消息hello work.queue-38 时间2025-02-05T14:53:04.793774100
***消费者2接收到消息hello work.queue-40 时间2025-02-05T14:53:05.006103400
***消费者2接收到消息hello work.queue-42 时间2025-02-05T14:53:05.220121400
***消费者2接收到消息hello work.queue-44 时间2025-02-05T14:53:05.433498300
***消费者2接收到消息hello work.queue-46 时间2025-02-05T14:53:05.645486500
***消费者2接收到消息hello work.queue-48 时间2025-02-05T14:53:05.856447600
***消费者2接收到消息hello work.queue-50 时间2025-02-05T14:53:06.065771700
可以看到消费者1和消费者2竟然每人消费了25条消息
消费者1很快完成了自己的25条消息消费者2却在缓慢的处理自己的25条消息。
也就是说消息是平均分配给每个消费者并没有考虑到消费者的处理能力。导致1个消费者空闲另一个消费者忙的不可开交。没有充分利用每一个消费者的能力最终消息处理的耗时远远超过了1秒。这样显然是有问题的。 能者多劳
更改一下我们的配置文件就好了。更改的是consumer消费者服务 application.yml 配置文件。
spring:rabbitmq:listener:simple:prefetch: 1 # 每次只能获取一条消息处理完成才能获取下一个消息 重启项目再次测试看结果。
消费者1接收到消息hello work.queue-1 时间2025-02-05T16:19:40.610672600
***消费者2接收到消息hello work.queue-2 时间2025-02-05T16:19:40.635078900
消费者1接收到消息hello work.queue-3 时间2025-02-05T16:19:40.668399800
消费者1接收到消息hello work.queue-4 时间2025-02-05T16:19:40.733468200
消费者1接收到消息hello work.queue-5 时间2025-02-05T16:19:40.789432700
消费者1接收到消息hello work.queue-6 时间2025-02-05T16:19:40.849740
***消费者2接收到消息hello work.queue-7 时间2025-02-05T16:19:40.865255600
消费者1接收到消息hello work.queue-8 时间2025-02-05T16:19:40.915186600
消费者1接收到消息hello work.queue-9 时间2025-02-05T16:19:40.975302
消费者1接收到消息hello work.queue-10 时间2025-02-05T16:19:41.035238100
消费者1接收到消息hello work.queue-11 时间2025-02-05T16:19:41.098149900
***消费者2接收到消息hello work.queue-12 时间2025-02-05T16:19:41.110162300
消费者1接收到消息hello work.queue-13 时间2025-02-05T16:19:41.158752
消费者1接收到消息hello work.queue-14 时间2025-02-05T16:19:41.214050800
消费者1接收到消息hello work.queue-15 时间2025-02-05T16:19:41.275456500
消费者1接收到消息hello work.queue-16 时间2025-02-05T16:19:41.338280900
***消费者2接收到消息hello work.queue-17 时间2025-02-05T16:19:41.354040400
消费者1接收到消息hello work.queue-18 时间2025-02-05T16:19:41.397333900
消费者1接收到消息hello work.queue-19 时间2025-02-05T16:19:41.459536100
消费者1接收到消息hello work.queue-20 时间2025-02-05T16:19:41.522984800
消费者1接收到消息hello work.queue-21 时间2025-02-05T16:19:41.589369900
***消费者2接收到消息hello work.queue-22 时间2025-02-05T16:19:41.595472400
消费者1接收到消息hello work.queue-23 时间2025-02-05T16:19:41.639076100
消费者1接收到消息hello work.queue-24 时间2025-02-05T16:19:41.702762100
消费者1接收到消息hello work.queue-25 时间2025-02-05T16:19:41.761438700
消费者1接收到消息hello work.queue-26 时间2025-02-05T16:19:41.823348300
***消费者2接收到消息hello work.queue-27 时间2025-02-05T16:19:41.836398700
消费者1接收到消息hello work.queue-28 时间2025-02-05T16:19:41.894946600
消费者1接收到消息hello work.queue-29 时间2025-02-05T16:19:41.962451900
消费者1接收到消息hello work.queue-30 时间2025-02-05T16:19:42.020227900
***消费者2接收到消息hello work.queue-31 时间2025-02-05T16:19:42.066749100
消费者1接收到消息hello work.queue-32 时间2025-02-05T16:19:42.080599800
消费者1接收到消息hello work.queue-33 时间2025-02-05T16:19:42.143280700
消费者1接收到消息hello work.queue-34 时间2025-02-05T16:19:42.204272700
消费者1接收到消息hello work.queue-35 时间2025-02-05T16:19:42.270407300
***消费者2接收到消息hello work.queue-36 时间2025-02-05T16:19:42.309818400
消费者1接收到消息hello work.queue-37 时间2025-02-05T16:19:42.332003100
消费者1接收到消息hello work.queue-38 时间2025-02-05T16:19:42.391974600
消费者1接收到消息hello work.queue-39 时间2025-02-05T16:19:42.454012300
消费者1接收到消息hello work.queue-40 时间2025-02-05T16:19:42.509398500
***消费者2接收到消息hello work.queue-41 时间2025-02-05T16:19:42.555230800
消费者1接收到消息hello work.queue-42 时间2025-02-05T16:19:42.570220
消费者1接收到消息hello work.queue-43 时间2025-02-05T16:19:42.629378200
消费者1接收到消息hello work.queue-44 时间2025-02-05T16:19:42.690519600
消费者1接收到消息hello work.queue-45 时间2025-02-05T16:19:42.756214500
***消费者2接收到消息hello work.queue-46 时间2025-02-05T16:19:42.797371400
消费者1接收到消息hello work.queue-47 时间2025-02-05T16:19:42.813034800
消费者1接收到消息hello work.queue-48 时间2025-02-05T16:19:42.876228100
消费者1接收到消息hello work.queue-49 时间2025-02-05T16:19:42.939391
消费者1接收到消息hello work.queue-50 时间2025-02-05T16:19:42.998590500 可以发现由于消费者1处理速度较快所以处理了更多的消息消费者2处理速度较慢只处理了7条消息。而最终总的执行耗时也在1秒左右大大提升。
正所谓能者多劳这样充分利用了每一个消费者的处理能力可以有效避免消息积压问题。 总结
Work模型的使用
多个消费者绑定到一个队列同一条消息只会被一个消费者处理通过设置prefetch来控制消费者预取的消息数量 交换机类型
在之前的两个测试案例中都没有交换机Exchange生产者直接发送消息到队列。而一旦引入交换机消息发送的模式会有很大变化 可以看到在订阅模型中多了一个exchange角色而且过程略有变化 Publisher生产者不再发送消息到队列中而是发给交换机Exchange交换机一方面接收生产者发送的消息。另一方面知道如何处理消息例如递交给某个特别队列、递交给所有队列、或是将消息丢弃。到底如何操作取决于Exchange的类型。Queue消息队列也与以前一样接收消息、缓存消息。不过队列一定要与交换机绑定。Consumer消费者与以前一样订阅队列没有变化 Exchange交换机只负责转发消息不具备存储消息的能力因此如果没有任何队列与Exchange绑定或者没有符合路由规则的队列那么消息会丢失
交换机的类型有四种 Fanout广播将消息交给所有绑定到交换机的队列。我们最早在控制台使用的正是Fanout交换机Direct订阅基于RoutingKey路由key发送给订阅了消息的队列Topic通配符订阅与Direct类似只不过RoutingKey可以使用通配符Headers头匹配基于MQ的消息头匹配用的较少 文档中我们讲解前面的三种交换机模式。 Fanout交换机
Fanout英文翻译是扇出我觉得在MQ中叫广播更合适。
在广播模式下消息发送流程是这样的 1 可以有多个队列2 每个队列都要绑定到Exchange交换机3 生产者发送的消息只能发送到交换机4 交换机把消息发送给绑定过的所有队列5 订阅队列的消费者都能拿到消息
我们的计划是这样的 创建一个名为test.fanout的交换机类型是Fanout创建两个队列fanout.queue1和fanout.queue2绑定到交换机test.fanout 声明交换机和队列
在控制台创建 fanout.queue1 和 fanout.queue2 两个队列。 然后在创建一个交换机 绑定两个队列到交换机 消息发送
在publisher服务的FanoutExchangeTest类中添加测试方法
SpringBootTest
public class FanoutExchangeTest {Autowiredprivate RabbitTemplate rabbitTemplate;/*测试 fanout exchange向 test.fanout 交换机发送消息消息内容为 hello everyone!会发送到所有绑定到该交换机的队列*/Testpublic void testSendFanoutExchange(){// 1 交换机名称String exchangeName test.fanout;// 2 消息String msg hello everyone;// 3 发送消息rabbitTemplate.convertAndSend(exchangeName,,msg);}
}
注意上述的 convertAndSend 方法的第2个参数路由key 因为没有绑定所以可以指定为空 看看rabbitMQ的控制台 消息接收
在consumer服务中添加FanoutQueueListener类并新增两个方法监听队列中的消息 作为消费者。
Slf4j
Component
public class FanoutQueueListener {/*** 监听fanout.queue1队列*/RabbitListener(queues fanout.queue1)public void listenFanoutQueue1(String msg){System.out.println(【消费者1】 接收到消息 msg);}/*** 监听fanout.queue2队列*/RabbitListener(queues fanout.queue2)public void listenFanoutQueue2(String msg){System.out.println(【消费者2】 接收到消息 msg);}
} 总结
交换机的作用是什么
接收publisher发送的消息将消息按照规则路由到与之绑定的队列不能缓存消息路由失败消息丢失FanoutExchange的会将消息路由到每个绑定的队列 Direct交换机
在Fanout模式中一条消息会被所有订阅的队列都消费。但是在某些场景下我们希望不同的消息被不同的队列消费。这时就要用到Direct类型的Exchange。 在Direct模型下
队列与交换机的绑定不能是任意绑定了而是要指定一个RoutingKey路由key消息的发送方在向 Exchange发送消息时也必须指定消息的 RoutingKey。Exchange不再把消息交给每一个绑定的队列而是根据消息的Routing Key进行判断只有队列的Routingkey与消息的 Routing key完全一致才会接收到消息。
案例需求如图 声明一个名为test.direct的交换机声明队列direct.queue1绑定hmall.directbindingKey为blud和red声明队列direct.queue2绑定hmall.directbindingKey为yellow和red在consumer服务中编写两个消费者方法分别监听direct.queue1和direct.queue2在publisher中编写测试方法向test.direct发送消息 声明队列和交换机
首先在控制台声明两个队列direct.queue1和direct.queue2这里不再展示过程 然后声明一个direct类型的交换机命名为test.direct: 然后使用red和blue作为key绑定direct.queue1到test.direct 绑定diretc.queue2 看看最后的绑定关系 消息发送
在publish服务中新增 DirectExchangeTest 类发送消息。
SpringBootTest
public class DirectExchangeTest {Autowiredprivate RabbitTemplate rabbitTemplate;/*测试 direct exchange向 test.direct 交换机发送消息会根据路由key发送到所有绑定到该交换机的队列*/Testpublic void testSendDirectExchange(){// 1 交换机String exchangeName test.direct;// 2 消息String msg 这是一条消息并且路由key是red 红色。;// 3 发送消息 路由key为redrabbitTemplate.convertAndSend(exchangeName,red,msg);//改变下消息msg 这是一条消息并且路由key是blue 蓝色。;rabbitTemplate.convertAndSend(exchangeName,blue,msg);}}
看看rabbitMQ控制台查看消息是否成功发送。 消息接收
在consumer服务中添加 DirectQueueListener 类并在里面编写两个方法。
Slf4j
Component
public class DirectQueueListener {/*** 监听direct.queue1队列*/RabbitListener(queues direct.queue1)public void listenDirectQueue1(String msg){log.info(【消费者1】接收到消息{},msg);}/*** 监听direct.queue2队列* param msg*/RabbitListener(queues direct.queue2)public void listenDirectQueue2(String msg){log.info(【消费者2】接收到消息{},msg);}
} 由于 test.redirect 交换机绑定的两个队列的路由key有red所以指定了路由key为red的消息能被两个消费者都收到。
而路由key为 blue 的队列只有direct.queue1所以只有监听这个队列的 消费者1 能够接收到消息 总结
描述下Direct交换机与Fanout交换机的差异
Fanout交换机将消息路由给每一个与之绑定的队列Direct交换机根据RoutingKey判断路由给哪个队列如果多个队列具有相同的RoutingKey则与Fanout功能类似 Topic交换机
Topic类型交换机
Topic类型的Exchange与Direct相比都是可以根据RoutingKey把消息路由到不同的队列。
只不过Topic类型Exchange可以让队列在绑定RoutingKey 的时候使用通配符
RoutingKey 一般都是有一个或多个单词组成多个单词之间以.分割例如 item.insert
通配符规则
#匹配一个或多个词*匹配不多不少恰好1个词
举例
item.#能够匹配item.spu.insert 或者 item.spuitem.*只能匹配item.spu
图示 假如此时publisher发送的消息使用的RoutingKey共有四种
china.news代表有中国的新闻消息china.weather 代表中国的天气消息japan.news 则代表日本新闻japan.weather 代表日本的天气消息
解释
topic.queue1绑定的是china.# 凡是以 china.开头的routing key 都会被匹配到包括 china.newschina.weather
topic.queue2绑定的是#.news 凡是以 .news结尾的 routing key 都会被匹配。包括: china.newsjapan.news
接下来我们就按照上图所示来演示一下Topic交换机的用法。
首先在控制台按照图示例子创建队列、交换机并利用通配符绑定队列和交换机。此处步骤略。最终结果如下 创建交换机和队列 创建test.topic 交换机 看看效果 给test.topic 交换机绑定两个队列 消息发送
在consumer服务中新增 TopicExchangeTest类 发送消息。 SpringBootTest
public class TopicExchangeTest {Autowiredprivate RabbitTemplate rabbitTemplate;Testpublic void testSendTopicExchange(){// 1 交换机String exchangeName test.topic;// 2 消息String msg 我是TopicExchange交换机的消息,路由key是 china.news;// 3 发送路由key为 china.news 的消息rabbitTemplate.convertAndSend(exchangeName,china.news,msg);}
} 消息接收
在consumer服务中添加 TopicExchangeListener 类编写两个方法监听消息。
Slf4j
Component
public class TopicExchangeListener {/*** 监听topic.queue1队列*/RabbitListener(queues topic.queue1)public void listenTopicQueue1(String msg) {log.info(【消费者1】监听到消息{}, msg);}/*** 监听topic.queue2队列*/RabbitListener(queues topic.queue2)public void listenTopicQueue2(String msg) {log.info(【消费者2】监听到消息{}, msg);}} 总结
描述下Direct交换机与Topic交换机的差异
Topic交换机接收的消息RoutingKey必须是多个单词以 . 分割Topic交换机与队列绑定时的RoutingKey可以指定通配符#代表0个或多个词*代表1个词 代码声明交换机和队列
在之前我们都是基于RabbitMQ控制台来创建队列、交换机。但是在实际开发时队列和交换机是程序员定义的将来项目上线又要交给运维去创建。那么程序员就需要把程序中运行的所有队列和交换机都写下来交给运维。在这个过程中是很容易出现错误的。
因此推荐的做法是由程序启动时检查队列和交换机是否存在如果不存在自动创建。 基本API
SpringAMQP提供了一个Queue类用来创建队列 SpringAMQP还提供了一个Exchange接口来表示所有不同类型的交换机 我们可以自己创建队列和交换机不过SpringAMQP还提供了ExchangeBuilder来简化这个过程 而在绑定队列和交换机时则需要使用BindingBuilder来创建Binding对象 把之前创建的队列和交换机删除
删除后的队列 删除后的交换机 Ideal控制台报错 这是因为我们的队列和交换机都删除了里面写的 RabbitListener 还在监听队列中的消息但是队列没有了所以报错。
fanout示例 在consumer服务中新建config包。并创建FanoutConfig 类 在里面编写代码创建test.fanout 交换机和fanout.queue1 和fanout.queue2 队列。 并启动consumer服务 Configuration
public class FanoutConfig {// 声明 Fanout 类型的交换机Beanpublic FanoutExchange fanoutExchange(){return new FanoutExchange(test.fanout);}//声明队列名称为 fanout.queue1Beanpublic Queue fanoutQueue1(){return new Queue(fanout.queue1);}//绑定队列和交换机Beanpublic Binding fanoutBinding1(FanoutExchange fanoutExchange,Queue fanoutQueue1){return BindingBuilder.bind(fanoutQueue1).to(fanoutExchange);}//声明队列名称为 fanout.queue2Beanpublic Queue fanoutQueue2(){return new Queue(fanout.queue2);}//绑定队列和交换机Beanpublic Binding fanoutBinding2(FanoutExchange fanoutExchange,Queue fanoutQueue2){return BindingBuilder.bind(fanoutQueue2).to(fanoutExchange);}
}
看看rbbitMQ控制台效果 看看交换机 Direct示例 在consumer 服务中的 config包中新建 DirectConfig 类编写代码创建交换机和队列。direct模式由于要绑定多个key会比较麻烦一点因为每一个key都要写一个binding方法。 Configuration
public class DirectConfig {//声明 test.direct 交换机Beanpublic DirectExchange directExchange(){return new DirectExchange(test.direct);}//声明 direct.queue1 队列Beanpublic Queue directQueue1(){return new Queue(direct.queue1);}//绑定 direct.queue1 队列到 test.direct 交换机上 路由key是 redBeanpublic Binding directBindingQueue1Red(DirectExchange directExchange,Queue directQueue1){return BindingBuilder.bind(directQueue1).to(directExchange).with(red);}//绑定 direct.queue1 队列到 test.direct 交换机上 路由key是 blueBeanpublic Binding directBindingQueue1Blue(DirectExchange directExchange,Queue directQueue1){return BindingBuilder.bind(directQueue1).to(directExchange).with(blue);}//声明 direct.queue2 队列Beanpublic Queue directQueue2(){return new Queue(direct.queue2);}//绑定 direct.queue2 队列到 test.direct 交换机上 路由key是 redBeanpublic Binding directBindingQueue2Red(DirectExchange directExchange, Queue directQueue2){return BindingBuilder.bind(directQueue2).to(directExchange).with(red);}//绑定 direct.queue2 队列到 test.direct 交换机上 路由key是 yellowBeanpublic Binding directBindingQueue2Yellow(DirectExchange directExchange,Queue directQueue2){return BindingBuilder.bind(directQueue2).to(directExchange).with(yellow);}}
看看rabbitMQ控制台 看看交换机和绑定关系 Topic示例 在consumer 服务中的config包里面新创建TopicConfig类编写代码创建交换机和队列。 Configuration
public class TopicConfig {//声明 test.topic 交换机Beanpublic TopicExchange topicExchange(){return new TopicExchange(test.topic);}//声明 topic.queue1 队列Beanpublic Queue topicQueue1(){return new Queue(topic.queue1);}//绑定队列和交换机 路由key是 china.#Beanpublic Binding topicBinding1(TopicExchange topicExchange,Queue topicQueue1){return BindingBuilder.bind(topicQueue1).to(topicExchange).with(china.#);}//声明 topic.queue2 队列Beanpublic Queue topicQueue2(){return new Queue(topic.queue2);}//绑定队列和交换机 路由key是 #.newsBeanpublic Binding topicBinding2(TopicExchange topicExchange,Queue topicQueue2){return BindingBuilder.bind(topicQueue2).to(topicExchange).with(#.news);}
}
看看控制台效果 交换机 基于注解声明
基于Bean的方式声明队列和交换机比较麻烦Spring还提供了基于注解方式来声明。不过是在消息监听的时候基于注解的方式来声明。
例如我们同样声明Direct模式的交换机和队列用注解的方式声明下。 先把之前创建的 交换机和队列删除。 删除后的效果 Fanout示例
Configuration
public class FanoutRabbitListener {// 监听fanout.queue1 队列的消息RabbitListener(bindings QueueBinding(value Queue(fanout.queue1),exchange Exchange(value test.fanout,type ExchangeTypes.FANOUT),key ))public void listenFanoutQueue1(String msg){System.out.println(【消费者1】 监听到消息 msg);}// 监听fanout.queue2 队列的消息RabbitListener(bindings QueueBinding(value Queue(fanout.queue2),exchange Exchange(value test.fanout,type ExchangeTypes.FANOUT),key ))public void listenFanoutQueue2(String msg){System.out.println(【消费者2】 监听到消息 msg);}
}
启动consumer服务看效果 交换机和绑定关系 Direct示例 新建 DirectRabbitListener 类并在里面编写代码进行测试。 Configuration
public class DirectRabbitListener {// 声明 direct.queue1RabbitListener(bindings QueueBinding(value Queue(direct.queue1),exchange Exchange(value test.direct,type ExchangeTypes.DIRECT),key {red,blue}))public void listenDirectQueue1(String msg){System.out.println(【消费者1】 接收到消息 msg);}// 声明direct.queue2RabbitListener(bindings QueueBinding(value Queue(direct.queue2),exchange Exchange(value test.direct,type ExchangeTypes.DIRECT),key {red,yellow}))public void listenDirectQueue2(String msg){System.out.println(【消费者2】 接收到消息 msg);}
}
看看效果 交换机和绑定关系 Topic示例
在consumer服务中的 config包里面创建TopicRabbitListener类。编写代码进行测试 Configuration
public class TopicRabbitListener {//声明topic.queue1 队列RabbitListener(bindings QueueBinding(value Queue(topic.queue1),exchange Exchange(value test.topic,type ExchangeTypes.TOPIC),key {china.#}))public void listenTopicQueue1(String msg){System.out.println(【消费者1】接收到消息msg);}//声明 topic.queue2 队列RabbitListener(bindings QueueBinding(value Queue(topic.queue2),exchange Exchange(value test.topic,type ExchangeTypes.TOPIC),key {#.news}))public void listenTopicQueue2(String msg){System.out.println(【消费者2】接收到消息msg);}
}
看看效果 交换机和绑定关系
