RabbitMQ基础教程之基于配置的消费者实现
相关博文，推荐查看:
- RabbitMq基础教程之安装与测试
- RabbitMq基础教程之基本概念
- RabbitMQ基础教程之基本使用篇
- RabbitMQ基础教程之使用进阶篇
- RabbitMQ基础教程之Spring&JavaConfig使用篇
- RabbitMQ基础教程之Spring-JavaConfig-FactoryBean使用姿势
前面一篇介绍了使用工厂方式创建消费者，其中一个不太友好的地方就在配置都是硬编码的方式，不太灵活，那么是否可以结合前一篇的FactoryBean来实现从配置中来灵活的创建消费者呢？
I. 动态配置实现消费者程序
1. 配置文件加载
首先就是需要从配置文件中获取相应的配置信息，借助JavaConfig，加一个注解即可
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
| @Configuration
@PropertySource("classpath:dynamicConfig.properties")
public class DynSpringConfig {
@Autowired
private Environment environment;
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setHost(environment.getProperty("dyn.mq.host"));
factory.setPort(Integer.parseInt(environment.getProperty("dyn.mq.port")));
factory.setUsername(environment.getProperty("dyn.mq.uname"));
factory.setPassword(environment.getProperty("dyn.mq.pwd"));
factory.setVirtualHost(environment.getProperty("dyn.mq.vhost"));
return factory;
}
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
}
|
主要就是 @PropertySource("classpath:dynamicConfig.properties") ， 表示从dynamicConfig.properties文件中读取相应的配置，而这些配置，会存放在 Environment 容器内；
获取配置的方式，就是通过org.springframework.core.env.PropertyResolver#getProperty(java.lang.String)获取
2. 消费者通用实现
实现一个简单的通用的消费端，主要根据前一篇博文中定义的MQContainerFactory，来生成SimpleMessageListenerContainer，然后注入消费服务，并启动容器
1
2
3
4
5
6
7
8
9
10
11
12
13
14
| public class DynamicConsumer {
public DynamicConsumer(MQContainerFactory fac) throws Exception {
SimpleMessageListenerContainer container = fac.getObject();
container.setMessageListener(new AbsMQConsumer() {
@Override
public boolean process(Message message, Channel channel) {
System.out.println("DynamicConsumer: " + fac.getQueue() + " | " + new String(message.getBody()));
return true;
}
});
container.start();
}
}
|
上面是一个非常简单的实现，针对常见的的RabbitMQ消息消费而言，也可以写一个泛型类，然后借助Spring的事件机制，实现一个通用的消费端，一种case如下:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
| public class JsonMsgConsumer {
public JsonMsgConsumer(ApplicationContext apc, MQContainerFactory fac, Class<?> msgType) throws Exception {
SimpleMessageListenerContainer container = fac.getObject();
container.setMessageListener(new AbsMQConsumer() {
@Override
public boolean process(Message message, Channel channel) {
System.out.println("DynamicConsumer: " + fac.getQueue() + " | " + new String(message.getBody()));
Object type = JSONObject.parseObject(message.getBody(), msgType);
apc.publishEvent(type);
return true;
}
});
container.start();
}
}
|
如果message中的数据，是通过Json序列化方式存入，则使用方，只需要监听对应的Event消费数据即可，完全不用再关系消费端的情况了
3. MQContainerFactory 初始化
根据配置文件中的信息，初始化factory，这个可谓是最关键的地方了，实现也和之前大致类似，只不过是将硬编码改成配置信息读取而已，完整的配置文件如下
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
| @Configuration
@PropertySource("classpath:dynamicConfig.properties")
public class DynSpringConfig {
@Autowired
private Environment environment;
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setHost(environment.getProperty("dyn.mq.host"));
factory.setPort(Integer.parseInt(environment.getProperty("dyn.mq.port")));
factory.setUsername(environment.getProperty("dyn.mq.uname"));
factory.setPassword(environment.getProperty("dyn.mq.pwd"));
factory.setVirtualHost(environment.getProperty("dyn.mq.vhost"));
return factory;
}
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
@Bean
public AmqpProducer amqpProducer() {
return new AmqpProducer();
}
@Bean
public DynamicConsumer dynamicConsumer(ConnectionFactory connectionFactory, RabbitAdmin rabbitAdmin)
throws Exception {
MQContainerFactory fac = MQContainerFactory.builder().directExchange(environment.getProperty("dyn.mq.exchange"))
.queue(environment.getProperty("dyn.mq.queue"))
.autoDeleted(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoDeleted")))
.autoAck(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoAck")))
.durable(Boolean.parseBoolean(environment.getProperty("dyn.mq.durable")))
.routingKey(environment.getProperty("dyn.mq.routingKey")).rabbitAdmin(rabbitAdmin)
.connectionFactory(connectionFactory).build();
return new DynamicConsumer(fac);
}
}
|
4. 测试
配置文件内容：
1
2
3
4
5
6
7
8
9
10
11
| dyn.mq.host=127.0.0.1
dyn.mq.port=5672
dyn.mq.uname=admin
dyn.mq.pwd=admin
dyn.mq.vhost=/
dyn.mq.exchange=fac.direct.exchange
dyn.mq.queue=dyn.queue
dyn.mq.durable=true
dyn.mq.autoDeleted=false
dyn.mq.autoAck=false
dyn.mq.routingKey=fac-routing
|
测试方法
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
| @RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = DynSpringConfig.class)
public class DynamicConsumerUnit {
@Autowired
private AmqpProducer amqpProducer;
@Test
public void testDirectConsumer() throws InterruptedException {
String[] routingKey = new String[]{"hello.world", "fac-routing", "test1"};
for (int i = 0; i < 100; i++) {
amqpProducer.publishMsg("fac.direct.exchange", routingKey[i % 3],
">>> hello " + routingKey[i % 3] + ">>> " + i);
}
System.out.println("-------over---------");
Thread.sleep(1000 * 60 * 10);
}
}
|
执行之后，就可以看到正常的消费了
5. 扩充与小结
看完之后，可能有一个问题，为什么要这样做，好处是什么？
大部分的时候，从MQ获取消息的逻辑都一样，唯一的区别在于获取到数据之后做的业务而言，如果把这一块完全的抽象出来，通过配置的方式，那么额外的新增mq的消费，就不需要再改消费端的代码了，然后就会有一个疑问，上面的配置文件中，生成dynamicConsumer的bean不也是需要额外写么？
如果将配置信息，以某种数组的方式定义，遍历读取这些配置，然后创建多个DynamicConsuer实例，是否就能支持动态扩展呢？
将配置改成下面的进行尝试
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
| @Configuration
@PropertySource("classpath:dynamicConfig.properties")
public class DynSpringConfig {
@Autowired
private Environment environment;
@Bean
public ConnectionFactory connectionFactory() {
CachingConnectionFactory factory = new CachingConnectionFactory();
factory.setHost(environment.getProperty("dyn.mq.host"));
factory.setPort(Integer.parseInt(environment.getProperty("dyn.mq.port")));
factory.setUsername(environment.getProperty("dyn.mq.uname"));
factory.setPassword(environment.getProperty("dyn.mq.pwd"));
factory.setVirtualHost(environment.getProperty("dyn.mq.vhost"));
return factory;
}
@Bean
public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
return new RabbitAdmin(connectionFactory);
}
@Bean
public AmqpProducer amqpProducer() {
return new AmqpProducer();
}
@Autowired
private ConnectionFactory connectionFactory;
@PostConstruct
public void dynamicConsumer()
throws Exception {
RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
MQContainerFactory fac = MQContainerFactory.builder().directExchange(environment.getProperty("dyn.mq.exchange"))
.queue(environment.getProperty("dyn.mq.queue"))
.autoDeleted(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoDeleted")))
.autoAck(Boolean.parseBoolean(environment.getProperty("dyn.mq.autoAck")))
.durable(Boolean.parseBoolean(environment.getProperty("dyn.mq.durable")))
.routingKey(environment.getProperty("dyn.mq.routingKey")).rabbitAdmin(rabbitAdmin)
.connectionFactory(connectionFactory).build();
new DynamicConsumer(fac);
}
}
|
注意之前 dynamicConsumer 是bean的创建，改成了初始化一个实例，如果配置文件是数组，内部用一个遍历就可以全部加载，现在就需要验证上面的配置改动之后，是否依然可以消费数据
实测ok，部分输出如下
1
2
3
4
| -------over---------
DynamicConsumer: dyn.queue | >>> hello fac-routing>>> 1
DynamicConsumer: dyn.queue | >>> hello fac-routing>>> 4
DynamicConsumer: dyn.queue | >>> hello fac-routing>>> 7
|
II. 其他
项目地址
一灰灰的个人博客，记录所有学习和工作中的博文，欢迎大家前去逛逛
声明
尽信书则不如，已上内容，纯属一家之言，因个人能力有限，难免有疏漏和错误之处，如发现bug或者有更好的建议，欢迎批评指正，不吝感激
扫描关注
