以前看过的关于RabbitMQ核心消息模式的文章都是基于Java API的,最近看了下官方文档,发现这些核心消息模式都可以通过Spring AMQP来实现。于是总结了下RabbitMQ的核心知识点,包括RabbitMQ在Windows和Linux下的安装、5种核心消息模式的Spring AMQP实现,相信对于想要学习和回顾RabbitMQ的朋友都会有所帮助。
简介
RabbitMQ是最受欢迎的开源消息中间件之一,在全球范围内被广泛应用。RabbitMQ是轻量级且易于部署的,能支持多种消息协议。RabbitMQ可以部署在分布式系统中,以满足大规模、高可用的要求。
相关概念
我们先来了解下RabbitMQ中的相关概念,这里以5种消息模式中的路由模式为例。
标志 | 中文名 | 英文名 | 描述 |
---|---|---|---|
P | 生产者 | Producer | 消息的发送者,可以将消息发送到交换机 |
C | 消费者 | Consumer | 消息的接收者,从队列中获取消息并进行消费 |
X | 交换机 | Exchange | 接收生产者发送的消息,并根据路由键发送给指定队列 |
Q | 队列 | Queue | 存储从交换机发来的消息 |
type | 交换机类型 | type | 不同类型的交换机转发消息方式不同 |
fanout | 发布/订阅模式 | fanout | 广播消息给所有绑定交换机的队列 |
direct | 路由模式 | direct | 根据路由键发送消息 |
topic | 通配符模式 | topic | 根据路由键的匹配规则发送消息 |
5种消息模式
这5种消息模式是构建基于RabbitMQ的消息应用的基础,一定要牢牢掌握它们。学过RabbitMQ的朋友应该了解过这些消息模式的Java实现,这里我们使用Spring AMQP的形式来实现它们。
简单模式
简单模式是最简单的消息模式,它包含一个生产者、一个消费者和一个队列。生产者向队列里发送消息,消费者从队列中获取消息并消费。
模式示意图
Spring AMQP实现
首先需要在pom.xml中添加Spring AMQP的相关依赖;
<!--Spring AMQP依赖--> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency
然后修改application.yml,添加RabbitMQ的相关配置;
spring: rabbitmq: host: localhost port: 5672 virtual-host: /mall username: mall password: mall publisher-confirms: true #消息发送到交换器确认 publisher-returns: true #消息发送到队列确认
- 添加简单模式相关Java配置,创建一个名为simple.hello的队列、一个生产者和一个消费者;
@Configuration
public class SimpleRabbitConfig {
@Bean
public Queue hello() {
return new Queue("simple.hello");
}
@Bean
public SimpleSender simpleSender(){
return new SimpleSender();
}
@Bean
public SimpleReceiver simpleReceiver(){
return new SimpleReceiver();
}
}
生产者通过send方法向队列simple.hello中发送消息;
public class SimpleSender { private static final Logger LOGGER = LoggerFactory.getLogger(SimpleSender.class); @Autowired private RabbitTemplate template; private static final String queueName="simple.hello"; public void send() { String message = "Hello World!"; this.template.convertAndSend(queueName, message); LOGGER.info(" [x] Sent '{}'", message); } }
消费者从队列simple.hello中获取消息;
@RabbitListener(queues = "simple.hello") public class SimpleReceiver { private static final Logger LOGGER = LoggerFactory.getLogger(SimpleReceiver.class); @RabbitHandler public void receive(String in) { LOGGER.info(" [x] Received '{}'", in); } }
在controller中添加测试接口,调用该接口开始发送消息;
@Api(tags = "RabbitController", description = "RabbitMQ功能测试") @Controller @RequestMapping("/rabbit") public class RabbitController { @Autowired private SimpleSender simpleSender; @ApiOperation("简单模式") @RequestMapping(value = "/simple", method = RequestMethod.GET) @ResponseBody public CommonResult simpleTest() { for(int i=0;i<10;i++){ simpleSender.send(); ThreadUtil.sleep(1000); } return CommonResult.success(null); } }
- 运行后结果如下,可以发现生产者往队列中发送消息,消费者从队列中获取消息并消费。
工作模式
工作模式是指向多个互相竞争的消费者发送消息的模式,它包含一个生产者、两个消费者和一个队列。两个消费者同时绑定到一个队列上去,当消费者获取消息处理耗时任务时,空闲的消费者从队列中获取并消费消息。
模式示意图
- 添加工作模式相关Java配置,创建一个名为work.hello的队列、一个生产者和两个消费者;
@Configuration
public class WorkRabbitConfig {
@Bean
public Queue workQueue() {
return new Queue("work.hello");
}
@Bean
public WorkReceiver workReceiver1() {
return new WorkReceiver(1);
}
@Bean
public WorkReceiver workReceiver2() {
return new WorkReceiver(2);
}
@Bean
public WorkSender workSender() {
return new WorkSender();
}
}
- 生产者通过send方法向队列work.hello中发送消息,消息中包含一定数量的.号;
public class WorkSender {
private static final Logger LOGGER = LoggerFactory.getLogger(WorkSender.class);
@Autowired
private RabbitTemplate template;
private static final String queueName = "work.hello";
public void send(int index) {
StringBuilder builder = new StringBuilder("Hello");
int limitIndex = index % 3+1;
for (int i = 0; i < limitIndex; i++) {
builder.append('.');
}
builder.append(index+1);
String message = builder.toString();
template.convertAndSend(queueName, message);
LOGGER.info(" [x] Sent '{}'", message);
}
}
- 两个消费者从队列work.hello中获取消息,名称分别为instance 1和instance 2,消息中包含.号越多,耗时越长;
@RabbitListener(queues = "work.hello")
public class WorkReceiver {
private static final Logger LOGGER = LoggerFactory.getLogger(WorkReceiver.class);
private final int instance;
public WorkReceiver(int i) {
this.instance = i;
}
@RabbitHandler
public void receive(String in) {
StopWatch watch = new StopWatch();
watch.start();
LOGGER.info("instance {} [x] Received '{}'", this.instance, in);
doWork(in);
watch.stop();
LOGGER.info("instance {} [x] Done in {}s", this.instance, watch.getTotalTimeSeconds());
}
private void doWork(String in) {
for (char ch : in.toCharArray()) {
if (ch == '.') {
ThreadUtil.sleep(1000);
}
}
}
}
- 在controller中添加测试接口,调用该接口开始发送消息;
@Api(tags = "RabbitController", description = "RabbitMQ功能测试")
@Controller
@RequestMapping("/rabbit")
public class RabbitController {
@Autowired
private WorkSender workSender;
@ApiOperation("工作模式")
@RequestMapping(value = "/work", method = RequestMethod.GET)
@ResponseBody
public CommonResult workTest() {
for(int i=0;i<10;i++){
workSender.send(i);
ThreadUtil.sleep(1000);
}
return CommonResult.success(null);
}
}
- 运行后结果如下,可以发现生产者往队列中发送包含不同数量.号的消息,instance 1和instance 2消费者互相竞争,分别消费了一部分消息。
发布/订阅模式
发布/订阅模式是指同时向多个消费者发送消息的模式(类似广播的形式),它包含一个生产者、两个消费者、两个队列和一个交换机。两个消费者同时绑定到不同的队列上去,两个队列绑定到交换机上去,生产者通过发送消息到交换机,所有消费者接收并消费消息。
模式示意图
Spring AMQP实现
添加发布/订阅模式相关Java配置,创建一个名为exchange.fanout的交换机、一个生产者、两个消费者和两个匿名队列,将两个匿名队列都绑定到交换机;
@Configuration public class FanoutRabbitConfig { @Bean public FanoutExchange fanout() { return new FanoutExchange("exchange.fanout"); } @Bean public Queue fanoutQueue1() { return new AnonymousQueue(); } @Bean public Queue fanoutQueue2() { return new AnonymousQueue(); } @Bean public Binding fanoutBinding1(FanoutExchange fanout, Queue fanoutQueue1) { return BindingBuilder.bind(fanoutQueue1).to(fanout); } @Bean public Binding fanoutBinding2(FanoutExchange fanout, Queue fanoutQueue2) { return BindingBuilder.bind(fanoutQueue2).to(fanout); } @Bean public FanoutReceiver fanoutReceiver() { return new FanoutReceiver(); } @Bean public FanoutSender fanoutSender() { return new FanoutSender(); } }
添加发布/订阅模式相关Java配置,创建一个名为exchange.fanout的交换机、一个生产者、两个消费者和两个匿名队列,将两个匿名队列都绑定到交换机;
public class FanoutSender { private static final Logger LOGGER = LoggerFactory.getLogger(FanoutSender.class); @Autowired private RabbitTemplate template; private static final String exchangeName = "exchange.fanout"; public void send(int index) { StringBuilder builder = new StringBuilder("Hello"); int limitIndex = index % 3 + 1; for (int i = 0; i < limitIndex; i++) { builder.append('.'); } builder.append(index + 1); String message = builder.toString(); template.convertAndSend(exchangeName, "", message); LOGGER.info(" [x] Sent '{}'", message); } }
消费者从绑定的匿名队列中获取消息,消息中包含.号越多,耗时越长,由于该消费者可以从两个队列中获取并消费消息,可以看做两个消费者,名称分别为instance 1和instance 2
public class FanoutReceiver { private static final Logger LOGGER = LoggerFactory.getLogger(FanoutReceiver.class); @RabbitListener(queues = "#{fanoutQueue1.name}") public void receive1(String in) { receive(in, 1); } @RabbitListener(queues = "#{fanoutQueue2.name}") public void receive2(String in) { receive(in, 2); } private void receive(String in, int receiver) { StopWatch watch = new StopWatch(); watch.start(); LOGGER.info("instance {} [x] Received '{}'", receiver, in); doWork(in); watch.stop(); LOGGER.info("instance {} [x] Done in {}s", receiver, watch.getTotalTimeSeconds()); } private void doWork(String in) { for (char ch : in.toCharArray()) { if (ch == '.') { ThreadUtil.sleep(1000); } } } }
在controller中添加测试接口,调用该接口开始发送消息;
@Api(tags = "RabbitController", description = "RabbitMQ功能测试") @Controller @RequestMapping("/rabbit") public class RabbitController { @Autowired private FanoutSender fanoutSender; @ApiOperation("发布/订阅模式") @RequestMapping(value = "/fanout", method = RequestMethod.GET) @ResponseBody public CommonResult fanoutTest() { for(int i=0;i<10;i++){ fanoutSender.send(i); ThreadUtil.sleep(1000); } return CommonResult.success(null); } }
- 运行后结果如下,可以发现生产者往队列中发送包含不同数量.号的消息,instance 1和instance 2同时获取并消费了消息。
路由模式
路由模式是可以根据路由键选择性给多个消费者发送消息的模式,它包含一个生产者、两个消费者、两个队列和一个交换机。两个消费者同时绑定到不同的队列上去,两个队列通过路由键绑定到交换机上去,生产者发送消息到交换机,交换机通过路由键转发到不同队列,队列绑定的消费者接收并消费消息。
模式示意图
Spring AMQP实现
添加路由模式相关Java配置,创建一个名为exchange.direct的交换机、一个生产者、两个消费者和两个匿名队列,队列通过路由键都绑定到交换机,队列1的路由键为orange和black,队列2的路由键为green和black;
@Configuration public class DirectRabbitConfig { @Bean public DirectExchange direct() { return new DirectExchange("exchange.direct"); } @Bean public Queue directQueue1() { return new AnonymousQueue(); } @Bean public Queue directQueue2() { return new AnonymousQueue(); } @Bean public Binding directBinding1a(DirectExchange direct, Queue directQueue1) { return BindingBuilder.bind(directQueue1).to(direct).with("orange"); } @Bean public Binding directBinding1b(DirectExchange direct, Queue directQueue1) { return BindingBuilder.bind(directQueue1).to(direct).with("black"); } @Bean public Binding directBinding2a(DirectExchange direct, Queue directQueue2) { return BindingBuilder.bind(directQueue2).to(direct).with("green"); } @Bean public Binding directBinding2b(DirectExchange direct, Queue directQueue2) { return BindingBuilder.bind(directQueue2).to(direct).with("black"); } @Bean public DirectReceiver receiver() { return new DirectReceiver(); } @Bean public DirectSender directSender() { return new DirectSender(); } }
生产者通过send方法向交换机exchange.direct中发送消息,发送时使用不同的路由键,根据路由键会被转发到不同的队列;
public class DirectSender { @Autowired private RabbitTemplate template; private static final String exchangeName = "exchange.direct"; private final String[] keys = {"orange", "black", "green"}; private static final Logger LOGGER = LoggerFactory.getLogger(DirectSender.class); public void send(int index) { StringBuilder builder = new StringBuilder("Hello to "); int limitIndex = index % 3; String key = keys[limitIndex]; builder.append(key).append(' '); builder.append(index+1); String message = builder.toString(); template.convertAndSend(exchangeName, key, message); LOGGER.info(" [x] Sent '{}'", message); } }
消费者从自己绑定的匿名队列中获取消息,由于该消费者可以从两个队列中获取并消费消息,可以看做两个消费者,名称分别为instance 1和instance 2;
public class DirectReceiver { private static final Logger LOGGER = LoggerFactory.getLogger(DirectReceiver.class); @RabbitListener(queues = "#{directQueue1.name}") public void receive1(String in){ receive(in, 1); } @RabbitListener(queues = "#{directQueue2.name}") public void receive2(String in){ receive(in, 2); } private void receive(String in, int receiver){ StopWatch watch = new StopWatch(); watch.start(); LOGGER.info("instance {} [x] Received '{}'", receiver, in); doWork(in); watch.stop(); LOGGER.info("instance {} [x] Done in {}s", receiver, watch.getTotalTimeSeconds()); } private void doWork(String in){ for (char ch : in.toCharArray()) { if (ch == '.') { ThreadUtil.sleep(1000); } } } }
在controller中添加测试接口,调用该接口开始发送消息;
@Api(tags = "RabbitController", description = "RabbitMQ功能测试") @Controller @RequestMapping("/rabbit") public class RabbitController { @Autowired private DirectSender directSender; @ApiOperation("路由模式") @RequestMapping(value = "/direct", method = RequestMethod.GET) @ResponseBody public CommonResult directTest() { for(int i=0;i<10;i++){ directSender.send(i); ThreadUtil.sleep(1000); } return CommonResult.success(null); } }
- 运行后结果如下,可以发现生产者往队列中发送包含不同路由键的消息,instance 1获取到了orange和black消息,instance 2获取到了green和black消息。
通配符模式
通配符模式是可以根据路由键匹配规则选择性给多个消费者发送消息的模式,它包含一个生产者、两个消费者、两个队列和一个交换机。两个消费者同时绑定到不同的队列上去,两个队列通过路由键匹配规则绑定到交换机上去,生产者发送消息到交换机,交换机通过路由键匹配规则转发到不同队列,队列绑定的消费者接收并消费消息。
特殊匹配符号
- '*':只能匹配一个单词;
'#':可以匹配零个或多个单词。
模式示意图Spring AMQP实现
添加通配符模式相关Java配置,创建一个名为exchange.topic的交换机、一个生产者、两个消费者和两个匿名队列,匹配.orange.和..rabbit发送到队列1,匹配lazy.#发送到队列2;
@Configuration public class TopicRabbitConfig { @Bean public TopicExchange topic() { return new TopicExchange("exchange.topic"); } @Bean public Queue topicQueue1() { return new AnonymousQueue(); } @Bean public Queue topicQueue2() { return new AnonymousQueue(); } @Bean public Binding topicBinding1a(TopicExchange topic, Queue topicQueue1) { return BindingBuilder.bind(topicQueue1).to(topic).with("*.orange.*"); } @Bean public Binding topicBinding1b(TopicExchange topic, Queue topicQueue1) { return BindingBuilder.bind(topicQueue1).to(topic).with("*.*.rabbit"); } @Bean public Binding topicBinding2a(TopicExchange topic, Queue topicQueue2) { return BindingBuilder.bind(topicQueue2).to(topic).with("lazy.#"); } @Bean public TopicReceiver topicReceiver() { return new TopicReceiver(); } @Bean public TopicSender topicSender() { return new TopicSender(); } }
生产者通过send方法向交换机exchange.topic中发送消息,消息中包含不同的路由键;
public class TopicSender { @Autowired private RabbitTemplate template; private static final String exchangeName = "exchange.topic"; private static final Logger LOGGER = LoggerFactory.getLogger(TopicSender.class); private final String[] keys = {"quick.orange.rabbit", "lazy.orange.elephant", "quick.orange.fox", "lazy.brown.fox", "lazy.pink.rabbit", "quick.brown.fox"}; public void send(int index) { StringBuilder builder = new StringBuilder("Hello to "); int limitIndex = index%keys.length; String key = keys[limitIndex]; builder.append(key).append(' '); builder.append(index+1); String message = builder.toString(); template.convertAndSend(exchangeName, key, message); LOGGER.info(" [x] Sent '{}'",message); System.out.println(" [x] Sent '" + message + "'"); } }
消费者从自己绑定的匿名队列中获取消息,由于该消费者可以从两个队列中获取并消费消息,可以看做两个消费者,名称分别为instance 1和instance 2;
public class TopicReceiver { private static final Logger LOGGER = LoggerFactory.getLogger(TopicReceiver.class); @RabbitListener(queues = "#{topicQueue1.name}") public void receive1(String in){ receive(in, 1); } @RabbitListener(queues = "#{topicQueue2.name}") public void receive2(String in){ receive(in, 2); } public void receive(String in, int receiver){ StopWatch watch = new StopWatch(); watch.start(); LOGGER.info("instance {} [x] Received '{}'", receiver, in); doWork(in); watch.stop(); LOGGER.info("instance {} [x] Done in {}s", receiver, watch.getTotalTimeSeconds()); } private void doWork(String in){ for (char ch : in.toCharArray()) { if (ch == '.') { ThreadUtil.sleep(1000); } } } }
在controller中添加测试接口,调用该接口开始发送消息;
@Api(tags = "RabbitController", description = "RabbitMQ功能测试") @Controller @RequestMapping("/rabbit") public class RabbitController { @Autowired private TopicSender topicSender; @ApiOperation("通配符模式") @RequestMapping(value = "/topic", method = RequestMethod.GET) @ResponseBody public CommonResult topicTest() { for(int i=0;i<10;i++){ topicSender.send(i); ThreadUtil.sleep(1000); } return CommonResult.success(null); } }
- 运行后结果如下,可以发现生产者往队列中发送包含不同路由键的消息,instance 1和instance 2分别获取到了匹配的消息。
评论 (0)