RabbitMQ基础教程之基于配置的消费者实现

相关博文,推荐查看:

  1. RabbitMq基础教程之安装与测试
  2. RabbitMq基础教程之基本概念
  3. RabbitMQ基础教程之基本使用篇
  4. RabbitMQ基础教程之使用进阶篇
  5. RabbitMQ基础教程之Spring&JavaConfig使用篇
  6. 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. 其他

项目地址

一灰灰Bloghttps://liuyueyi.github.io/hexblog

一灰灰的个人博客,记录所有学习和工作中的博文,欢迎大家前去逛逛

声明

尽信书则不如,已上内容,纯属一家之言,因个人能力有限,难免有疏漏和错误之处,如发现bug或者有更好的建议,欢迎批评指正,不吝感激

扫描关注

QrCode