摘要：需要對外提供配置接口，通過配置接口，動態(tài)配置，可接收的消息。遇到問題正式這種注解的方式，使得只能在代碼中寫死，沒法動態(tài)修改。而我想實(shí)現(xiàn)的效果是能夠動態(tài)的修改，動態(tài)的創(chuàng)建。因此能不能動態(tài)修改配置文件對應(yīng)的變量，然后消費(fèi)者動態(tài)注入。
背景
最近在寫一個(gè)Mqtt消息轉(zhuǎn)發(fā)的中間件,可通過ActvieMq接收消息。需要對外提供配置接口，通過配置接口，動態(tài)配置Queue，可接收Queue的消息。
整個(gè)項(xiàng)目依賴于SpringBoot ，通過SpringBoot實(shí)現(xiàn)隊(duì)列消費(fèi)，只需要通過@JmsListener(destination = "queueName") 注解，就可以實(shí)現(xiàn)對特定隊(duì)列的消費(fèi)。
遇到問題
正式這種注解的方式，使得destination只能在代碼中寫死，沒法動態(tài)修改。 而我想實(shí)現(xiàn)的效果是能夠動態(tài)的修改destination，動態(tài)的創(chuàng)建Consumer。
解決方案
繼續(xù)使用@JmsListener，找到某種方式，能夠動態(tài)修改destination
不使用@JmsListener，通過ActiveMq提供的Jar，手動實(shí)現(xiàn)連接、消費(fèi)等。之前通過這種方式實(shí)現(xiàn)過RabbitMq的處理，因此該方案不會有難度，只是工作量多一些。
由于SpringBoot的過分簡單，因此開始嘗試通過第一種方式。
解決過程
一開始想著通過反射修改注解destination但是嘗試失敗。后來想到@JmsListener(destination = "${xxx}")這種方式，根據(jù)配置文件，可以修改消費(fèi)的隊(duì)列，但是需要從新啟動。 因此能不能動態(tài)修改配置文件對應(yīng)的變量，然后消費(fèi)者動態(tài)注入Spring。
自定義實(shí)現(xiàn)一個(gè)MapPropertySource，然后加入到Environment中
@Configuration
public class DynamicTestSetting {
public static final String DYNAMIC_CONFIG = "dynamic_settting";
@Autowired
AbstractEnvironment environment;
@PostConstruct
public void init() {
environment.getPropertySources().addFirst(new DynamicLoadPropertySource(DYNAMIC_CONFIG, null));
}
}
@Slf4j
public class DynamicLoadPropertySource extends MapPropertySource {
private static Map map = new ConcurrentHashMap(64);
private static ScheduledExecutorService scheduled = Executors.newScheduledThreadPool(1);
static {
scheduled.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
//測試：定時(shí)修改value
//真正使用可能需要傳遞一個(gè)參數(shù)或者定時(shí)讀取配置文件
map.put("test", String.valueOf(System.currentTimeMillis()));
}
}, 1, 1, TimeUnit.SECONDS);
}
public DynamicLoadPropertySource(String name, Map source) {
super(name, map);
}
@Override
public Object getProperty(String name) {
return map.get(name);
}
}
2.消費(fèi)者需要?jiǎng)討B(tài)加入
@Slf4j
public class TestConsumer {
@JmsListener(destination = "${test}")
public void receiveQueue(BytesMessage msg) {
//接收消息后的處理邏輯
}
}
/*動態(tài)注入bean到spring，需要用到ApplicationContext/
DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) context.getAutowireCapableBeanFactory();
BeanDefinitionBuilder definitionBuilder = BeanDefinitionBuilder.genericBeanDefinition(TestConsumer.class);
defaultListableBeanFactory.registerBeanDefinition("testConsumer",definitionBuilder.getBeanDefinition());
//調(diào)用getBeaan時(shí)Spring會創(chuàng)建一個(gè)TestConsumer實(shí)例，這個(gè)時(shí)候ActiveMq中會創(chuàng)建一個(gè)destination = "${test}"隊(duì)列，TestConsummer和其綁定消費(fèi)
TestConsumer consumer = context.getBean("testConsumer", TestConsumer.class);
System.out.println(consumer);
動態(tài)關(guān)閉Queue的消費(fèi)者
參見另一篇：https://segmentfault.com/n/13...
參考文章：https://blog.csdn.net/qq_3491...
文章版權(quán)歸作者所有，未經(jīng)允許請勿轉(zhuǎn)載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。
轉(zhuǎn)載請注明本文地址：http://systransis.cn/yun/71482.html
