背景
最近在写一个Mqtt消息转发的中间件,可通过ActvieMq接收消息。需要对外提供配置接口,通过配置接口,动态配置Queue,可接收Queue的消息。
整个项目依赖于SpringBoot ,通过SpringBoot实现队列消费,只需要通过@JmsListener(destination = "queueName") 注解,就可以实现对特定队列的消费。
遇到问题
正式这种注解的方式,使得destination只能在代码中写死,没法动态修改。 而我想实现的效果是能够动态的修改destination,动态的创建Consumer。
解决方案
- 继续使用@JmsListener,找到某种方式,能够动态修改destination
- 不使用@JmsListener,通过ActiveMq提供的Jar,手动实现连接、消费等。之前通过这种方式实现过RabbitMq的处理,因此该方案不会有难度,只是工作量多一些。
由于SpringBoot的过分简单,因此开始尝试通过第一种方式。
解决过程
一开始想着通过反射修改注解destination但是尝试失败。后来想到@JmsListener(destination = "${xxx}")这种方式,根据配置文件,可以修改消费的队列,但是需要从新启动。 因此能不能动态修改配置文件对应的变量,然后消费者动态注入Spring。
- 自定义实现一个MapPropertySource,然后加入到Environment中
@Configuration
public class DynamicTestSetting {public static final String DYNAMIC_CONFIG = "dynamic_settting";@AutowiredAbstractEnvironment environment;@PostConstructpublic void init() { environment.getPropertySources().addFirst(new DynamicLoadPropertySource(DYNAMIC_CONFIG, null));}
}
@Slf4j
public class DynamicLoadPropertySource extends MapPropertySource {private static Mapmap = new ConcurrentHashMap (64);private static ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1);static { scheduled.scheduleAtFixedRate(new Runnable() { @Override public void run() { //测试:定时修改value //真正使用可能需要传递一个参数或者定时读取配置文件 map.put("test", String.valueOf(System.currentTimeMillis())); } }, 1, 1, TimeUnit.SECONDS);}public DynamicLoadPropertySource(String name, Map source) { super(name, map);}@Overridepublic Object getProperty(String name) { return map.get(name);}
}
2.消费者需要动态加入
@Slf4jpublic class TestConsumer {@JmsListener(destination = "${test}")public void receiveQueue(BytesMessage msg) { //接收消息后的处理逻辑}
}
/*动态注入bean到spring,需要用到ApplicationContext/
DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) context.getAutowireCapableBeanFactory();BeanDefinitionBuilder definitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(TestConsumer.class); defaultListableBeanFactory.registerBeanDefinition("testConsumer",definitionBuilder.getBeanDefinition());
//调用getBeaan时Spring会创建一个TestConsumer实例,这个时候ActiveMq中会创建一个destination = "${test}"队列,TestConsummer和其绑定消费
TestConsumer consumer = context.getBean("testConsumer", TestConsumer.class); System.out.println(consumer);
动态关闭Queue的消费者
参见另一篇:
参考文章: