背景
最近在写一个Mqtt消息转发的中间件,可经过ActvieMq接收消息。须要对外提供配置接口，经过配置接口，动态配置Queue，可接收Queue的消息。spring
整个项目依赖于SpringBoot ，经过SpringBoot实现队列消费，只须要经过@JmsListener(destination = "queueName") 注解，就能够实现对特定队列的消费。segmentfault
遇到问题
正式这种注解的方式，使得destination只能在代码中写死，无法动态修改。 而我想实现的效果是可以动态的修改destination，动态的建立Consumer。ide
解决方案
- 继续使用@JmsListener，找到某种方式，可以动态修改destination
- 不使用@JmsListener，经过ActiveMq提供的Jar，手动实现链接、消费等。以前经过这种方式实现过RabbitMq的处理，所以该方案不会有难度，只是工做量多一些。
因为SpringBoot的过度简单，所以开始尝试经过第一种方式。测试
解决过程
一开始想着经过反射修改注解destination可是尝试失败。后来想到@JmsListener(destination = "${xxx}")这种方式，根据配置文件，能够修改消费的队列，可是须要重新启动。 所以能不能动态修改配置文件对应的变量，而后消费者动态注入Spring。ui
- 自定义实现一个MapPropertySource，而后加入到Environment中
@Configuration
public class DynamicTestSetting {.net
public static final String DYNAMIC_CONFIG = "dynamic_settting";
@Autowired
AbstractEnvironment environment;
@PostConstruct
public void init() {
environment.getPropertySources().addFirst(new DynamicLoadPropertySource(DYNAMIC_CONFIG, null));
}
}code
@Slf4j
public class DynamicLoadPropertySource extends MapPropertySource {中间件
private static Map<String, Object> map = new ConcurrentHashMap<String, Object>(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<String, Object> source) {
super(name, map);
}
@Override
public Object getProperty(String name) {
return map.get(name);
}
} blog
2.消费者须要动态加入
@Slf4j
public 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);