国信证券 RocketMQ + SpringBoot 在配置热加载的实践

国信证券 RocketMQ + SpringBoot 在配置热加载的实践

在进行微服务架构研发的过程中,不少研发人员都提出配置热加载的需求,一方面是为了提升研发效率,开发人员修改配置不需要重启服务;另一方面是为了线上紧急事件,可以快速恢复,例如线上数据库 down 了,可以紧急启动降级开关,通过配置热加载动态生效,降低处理事件的时间。于是我采用 rocketmq 消息队列,实现一套解耦的配置热加载机制。这里我将思路简单介绍下,希望大家能通过这篇文章,实现自己的配置热加载。

1 设计方案

  1. 配置中心修改配置后,将修改的配置发送到 rocketmq。
  2. 微服务启动时,监控配置修改的topic,发现自身配置已修改,则发起配置热加载。
  3. 服务需要设置热加载开关,配置误修改可能带来风险。

1.1 RocketMQ消息设计

topic tag body
confCenter confModify 服务名称和配置内容

1.2 采用spring的event机制处理消息消息消费

RocketmqEvent 代码如下:

public class RocketmqEvent extends ApplicationEvent {
    private static final long serialVersionUID = -4468405250074063206L;
    private DefaultMQPushConsumer consumer;
    private List<MessageExt> msgs;

    public RocketmqEvent(List<MessageExt> msgs, DefaultMQPushConsumer consumer) throws Exception {
        super(msgs);
        this.consumer = consumer;
        this.setMsgs(msgs);
    }

    public String getMsg(int idx) {
        try {
            return new String(getMsgs().get(idx).getBody(), "utf-8");
        } catch (UnsupportedEncodingException e) {
            return null;
        }
    }

    public String getMsg(int idx,String code) {
        try {
            return new String(getMsgs().get(idx).getBody(), code);
        } catch (UnsupportedEncodingException e) {
            return null;
        }
    }

    public DefaultMQPushConsumer getConsumer() {
        return consumer;
    }

    public void setConsumer(DefaultMQPushConsumer consumer) {
        this.consumer = consumer;
    }

    public MessageExt getMessageExt(int idx) {
        return getMsgs().get(idx);
    }


    public String getTopic(int idx) {
        return getMsgs().get(idx).getTopic();
    }


    public String getTag(int idx) {
        return getMsgs().get(idx).getTags();
    }


    public byte[] getBody(int idx) {
        return getMsgs().get(idx).getBody();
    }


    public String getKeys(int idx) {
        return getMsgs().get(idx).getKeys();
    }

    public List<MessageExt> getMsgs() {
        return msgs;
    }

    public void setMsgs(List<MessageExt> msgs) {
        this.msgs = msgs;
    }
}

1.3 RocketMQ的message的Event处理

@Autowired
ApplicationEventPublisher publisher 
publisher.publishEvent(new RocketmqEvent(msgs, consumer));

2 spring配置热加载

微服务端收到消息后,采用spring的RefreshScope机制进行配置热加载
1、需要注入 RefreshScope 和 ConfigurableWebApplicationContext 两个对象。

ConfigurableWebApplicationContext applicationContext;\\采用ApplicationContextAware机制进行注入
@Autowired
private RefreshScope refreshScope = null;

2、解析 RocketMQ 的 message,然后进行配置刷新。

@EventListener
    public void envListener(RocketmqEvent event) {
        event.getMsgs().forEach(msg -> {
            String body = new String(msg.getBody());
            JSONObject json = JSON.parseObject(body);
            MutablePropertySources target = applicationContext.getEnvironment().getPropertySources();//获取Sources
            for (PropertySource<?> source : target) {
                if ("defaultProperties".equals(source.getName())) {//对默认配置进行刷新
                    @SuppressWarnings("unchecked")
                    Map<String, String> map = (Map<String, String>) source.getSource();
                    Map<String, String> after = JSONObject.parseObject(json.toJSONString(),
                            new TypeReference<Map<String, String>>() {
                            });
                    map.putAll(after);
                    PropertiesContent.refleshAll(after);
                    this.refreshScope.refreshAll();
                    return;
                }
            }
        });
    }

最后,这里只是简单介绍了配置热加载的思路,当然还有很多细节需要思考,如果有兴趣的可以一起交流交流。

作者简介:邓启翔,国信证券 zebra 微服务技术负责人,具有多年网络框架和分布式系统架构设计及研发经验,对消息中间件具有深刻理解。曾就职于平安、汇添富基金,资深架构师。


原文地址: RocketMQ官微
原文作者: 邓启翔

我觉得老哥可以来一个整合RocketMq的教程

有官方的 starter,而且还是中文的文档。你可以研究一下。

https://github.com/apache/rocketmq-spring/blob/master/README_zh_CN.md

大佬 你写的这个 配置中心 以及微服务 还有rocketmq 都在同一个服务上吧

可以不用在同一个服务上啊。通过MQ通信就是了。配置服务把配置修改事件消息推送给MQ。各个微服务消费这个配置修改消息,修改本地的配置数据。

嗯嗯 这我明白 ,我的意思是 作者写的这个 demo 没有体现出 微服务

1 个赞

你好,我有个疑问:
如果我消费失败了,怎么可以通知broker进行重复消费呢
如果通过回调的方式,可以通过ConsumeConcurrentlyStatus.RECONSUME_LATER这个去重复消费,但是写法不够优雅,与springboot整合的写法,题主的写法更加优雅,但是我不知道,如果失败了该怎么告诉broker然后重复消费呢