Preface
Spring Boot集成RabbitMQ, 其属性可直接通过application.yml中的spring.rabbitmq.*前缀配置.
Sprint Boot RabbitMQ的消费者默认是Fair dispatch, 即prefetch=1
为了方便调试, 我将所有Exchange与Queue设置为auto delete.
以下所有示例基于spring-boot-starter-amqp 1.4.4.RELEASE, 若在未来的版本中有更加优雅的使用方法可以给我留言哈.
Prerequisite
application.yml中先定义账户密码等连接信息, 确保连接成功后再进行下一步操作
spring:
rabbitmq:
host: localhost
username: chris
password: 123123
virtual-host: prontera
Main Concepts
org.springframework.amqp.core.Queue: 队列
org.springframework.amqp.core.Binding: 建立交换器与队列的绑定关系
org.springframework.amqp.core.DirectExchange: Direct交换器
org.springframework.amqp.core.TopicExchange: Topic交换器
org.springframework.amqp.core.FanoutExchange: Fanout交换器
org.springframework.amqp.support.converter.MessageConverter: 消息转换器, 如将Java类转换JSON类型发送至Broker, 从Broker处获取JSON消息转换为Java类型
org.springframework.amqp.core.AmqpTemplate 多用于生产者端发布消息
org.springframework.amqp.core.AmqpAdmin 用于Exchange, Queue等的动态管理
Configuration
Work queues
生产者与消费者的配置一样, 因为监听的是同一个队列, 所以队列名要先约定好
/**
* @author Zhao Junjian
*/
@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_DIRECT_EXCHANGE = "prontera.direct";
public static final String TRADE_QUUE = "funds";
public static final String TRADE_ROUTE_KEY = "trading";
@Bean
public DirectExchange pronteraExchange() {
return new DirectExchange(DEFAULT_DIRECT_EXCHANGE, true, true);
}
@Bean
public Queue tradeQueue() {
return new Queue(TRADE_QUEUE, true, false, true);
}
@Bean
public Binding tradeBinding() {
return BindingBuilder.bind(tradeQueue()).to(pronteraExchange()).with(TRADE_ROUTE_KEY);
}
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
}
生产消息, 这里我是借助Controller发起的请求, 无论怎样只要能注入AmqpTemplate就行
@Autowired
private AmqpTemplate amqpTemplate;
@RequestMapping(value = "/echo", method = RequestMethod.GET)
public Map<String, ?> hello() {
final WorkUnit unit = new WorkUnit();
unit.setId("1");
unit.setMessage("hello world");
amqpTemplate.convertAndSend(RabbitConfiguration.DEFAULT_DIRECT_EXCHANGE, RabbitConfiguration.TRADE_ROUTE_KEY, unit);
return ImmutableMap.of("code", 20000);
}
消费消息
@RabbitListener(queues = {RabbitConfiguration.TRADE_QUEUE})
public void processBootTask(WorkUnit content) {
System.out.println(content);
}
NOTE:
如果先启动生产者, 那么要先与RabbitMQ进行过通讯之后, 才会在RabbitMQ Management处看到Exchange与Queue. 如果是先启动消费者, 那么@RabbitListener会自动监听队列, 所以可以直接在RabbitMQ Management处看到我们所定义的组件
Publish/Subscribe
要使用Fanout的话那么两端的配置就不一样了, 根据AMQP规范, 生产者其实仅关注Exchange与Route Key, 消费者仅关注Queue, 根据这条规则我们来写一下生产端的配置
/**
* @author Zhao Junjian
*/
@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_FANOUT_EXCHANGE = "prontera.fanout";
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(DEFAULT_FANOUT_EXCHANGE, true, true);
}
}
消费端我们就需要将Exchange与临时Queue进行绑定, 这里我们使用UUID为Queue命名
/**
* @author Zhao Junjian
*/
@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_FANOUT_EXCHANGE = "prontera.fanout";
public static final String FANOUT_QUEUE = "p-" + UUID.randomUUID();
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public FanoutExchange fanoutExchange() {
return new FanoutExchange(DEFAULT_FANOUT_EXCHANGE, true, true);
}
@Bean
public Queue randomQueue() {
return new Queue(FANOUT_QUEUE, true, false, true);
}
@Bean
public Binding fanoutBinding() {
return BindingBuilder.bind(randomQueue()).to(fanoutExchange());
}
}
其实也可以使用UniquelyNamedQueue, 只不过我个人觉得没那么顺手而已
以下是生产端发送消息的示例, Route Key "chris"在Fanout Exchange中是会被忽略的, 这里只是提醒一下自己
@Autowired
private AmqpTemplate amqpTemplate;
@RequestMapping(value = "/echo", method = RequestMethod.GET)
public Map<String, ?> hello() {
// fanout
amqpTemplate.convertAndSend(RabbitConfiguration.DEFAULT_FANOUT_EXCHANGE, "chris", unit);
return ImmutableMap.of("code", 20000);
}
消费端获取消息
@RabbitListener(queues = "#{rabbitConfiguration.FANOUT_QUEUE}")
public void processBootTask(WorkUnit content) {
System.out.println(content);
}
需要注意的是, 因为注解中的值必须是常量, 如果我们直接写RabbitConfiguration.FANOUT_QUEUE是会抛出编译期的异常, 而Spring则为该注解进行SpEL扩展使其支持动态变量. 因为Topic与Fanout都必须使用临时队列(随机且唯一的对列名才有意义), 所以这里我们使用UUID为其命名.
Routing
与Work queues一样使用Direct Exchange, 就是消费者监听的队列不是同一个, 这里就不作演示了
Topics
生产端配置
/**
* @author Zhao Junjian
*/
@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_TOPIC_EXCHANGE = "prontera.topic";
public static final String TOPIC_ROUTE_KEY = "NYSE.TECH.MSFT";
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(DEFAULT_TOPIC_EXCHANGE, true, true);
}
}
消费端同样跟Fanout一样也是配置临时队列
/**
* @author Zhao Junjian
*/
@Configuration
public class RabbitConfiguration {
public static final String DEFAULT_TOPIC_EXCHANGE = "prontera.topic";
public static final String TOPIC_QUEUE = "p-" + UUID.randomUUID();
public static final String TOPIC_ROUTE_KEY = "#.#"; // 下表有例子
@Bean
public MessageConverter messageConverter() {
return new Jackson2JsonMessageConverter();
}
@Bean
public TopicExchange topicExchange() {
return new TopicExchange(DEFAULT_TOPIC_EXCHANGE, true, true);
}
@Bean
public Queue randomQueue() {
return new Queue(TOPIC_QUEUE, true, false, true);
}
@Bean
public Binding topicBinding() {
return BindingBuilder.bind(randomQueue()).to(topicExchange()).with(TOPIC_ROUTE_KEY);
}
}
生产端发送请求, 其实就是指定Exchange与Route Key, 对于Queue来说生产者不关心
@Autowired
private AmqpTemplate amqpTemplate;
@RequestMapping(value = "/echo", method = RequestMethod.GET)
public Map<String, ?> hello() {
final WorkUnit unit = new WorkUnit();
unit.setId("1");
unit.setMessage("hello world");
amqpTemplate.convertAndSend(RabbitConfiguration.DEFAULT_TOPIC_EXCHANGE, RabbitConfiguration.TOPIC_ROUTE_KEY, unit);
return ImmutableMap.of("code", 20000);
}
消费端跟Fanout中的例子是一样的
@RabbitListener(queues = "#{rabbitConfiguration.TOPIC_QUEUE}")
public void processBootTask(WorkUnit content) {
System.out.println(content);
}
生产端指定Route Key为NYSE.TECH.MSFT, 下面是消费端绑定Route Key的不同情况
| BINDING KEY ON CONSUMER SIDE | MATCH? |
|---|---|
| NYSE.TECH.MSFT | Yes |
| # | Yes |
| NYSE.# | Yes |
| . | No |
| NYSE.* | No |
| NYSE.TECH.* | Yes |
| NYSE.*.MSFT | Yes |
小结
本篇介绍了Spring Boot RabbitMQ的常用模型, 在下篇将会介绍prefetch, dead-letter与重试机制.
作者：Chris
原博客：http://blog.chriscs.com
