RocketMQ broker停写功能源码分析

偏执的太偏执、 2023-10-14 08:06 14阅读 0赞

## 背景 ##

在我们要平滑升级broker的时候，无损升级的最佳实践应该是

1.  新broker启动
2.  旧broker停写
3.  旧broker消息消费完成后下线(包括延时消息)

所以我们本次就是来分析如何完成broker的停写

## 源码入口 ##

其实通过查看源码，我们发现有两种方式可以停写broker

1.  通过mqadmin运维工具命令行的方式

![format_png][]

2.通过MQAdminExt管理工具类

![format_png 1][]

![get_code_ZGZkM2I1NGUwMTYzODRiM2E0ZjQ3NmMwZWEwZTkwMjQsMTY5MDQyNTE2ODQ0MA][]

两种方式没有区别，都是调用DefaultMQAdminExt的wipeWritePermOfBroker方法

![format_png 2][]

![get_code_ZDAyNDFmYjhhMGVkMzc2NzQ2NTUyNjU4NWJmNWU4NTksMTY5MDQyNTE2ODQ0MA][]

## 源码分析 ##

## client ##

我们这里直接进去到最底层的实现类代码

java复制代码public int wipeWritePermOfBroker(final String namesrvAddr, String brokerName,
            final long timeoutMillis) throws RemotingCommandException,
            RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException, MQClientException {
            WipeWritePermOfBrokerRequestHeader requestHeader = new WipeWritePermOfBrokerRequestHeader();
            requestHeader.setBrokerName(brokerName);
    
            RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.WIPE_WRITE_PERM_OF_BROKER, requestHeader);
            RemotingCommand response = this.remotingClient.invokeSync(namesrvAddr, request, timeoutMillis);
            assert response != null;
            switch (response.getCode()) {
                case ResponseCode.SUCCESS: {
                    WipeWritePermOfBrokerResponseHeader responseHeader =
                        (WipeWritePermOfBrokerResponseHeader) response.decodeCommandCustomHeader(WipeWritePermOfBrokerResponseHeader.class);
                    return responseHeader.getWipeTopicCount();
                }
                default:
                    break;
            }
    
            throw new MQClientException(response.getCode(), response.getRemark());
        }

客户端的代码我们没什么好看的，我们还是通过请求状态码RequestCode.WIPE\_WRITE\_PERM\_OF\_BROKER查看实际的Nameserver的处理逻辑

## NameServer ##

![format_png 3][]

![get_code_MzIxNmVlNGJiMzZmN2YwMTExMTdkNWM0YzdiMjM4MzcsMTY5MDQyNTE2ODQ0MA][]

*  DefaultRequestProcessor的方法 wipeWritePermOfBroker

可以看到实际的代码逻辑在这一行

java复制代码int wipeTopicCnt = this.namesrvController.getRouteInfoManager().wipeWritePermOfBrokerByLock(requestHeader.getBrokerName());

我们进入到这个方法看看

![format_png 4][]

![get_code_NDNlMjNkMGZhZjRiMTc5YTA0N2U3MTIyM2I1MmE3MjIsMTY5MDQyNTE2ODQ0MA][]

可以看到实际的逻辑还在更下层，我们再进去看看

java复制代码private int operateWritePermOfBroker(final String brokerName, final int requestCode) {
            int topicCnt = 0;
    
            for (Entry<String, Map<String, QueueData>> entry : this.topicQueueTable.entrySet()) {
                Map<String, QueueData> qdMap = entry.getValue();
    
                final QueueData qd = qdMap.get(brokerName);
                if (qd == null) {
                    continue;
                }
                int perm = qd.getPerm();
                switch (requestCode) {
                    case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
                        perm &= ~PermName.PERM_WRITE;
                        break;
                    case RequestCode.ADD_WRITE_PERM_OF_BROKER:
                        perm = PermName.PERM_READ | PermName.PERM_WRITE;
                        break;
                }
                qd.setPerm(perm);
                topicCnt++;
            }
            return topicCnt;
        }

这里的topicQueueTable我们可以看看他实际的数据结构

![format_png 5][]

![get_code_MmI1NzUxOGEyY2Y1NDI2NGFjNzNmMDQzMTdjOTA1OWUsMTY5MDQyNTE2ODQ0MA][]

可以看到这里他的perm是7

如果我们去broker的 config的topics.json查看相关的配置信息

可以看到刚好可以对得上

![format_png 6][]

![get_code_MGM2MzRjOGFlZDg5NmZiYzJjMzBiM2RmY2RhYmEzMTMsMTY5MDQyNTE2ODQ0MA][]

这里我们解释一下perm的权限问题

![format_png 7][]

![get_code_YWI1MWQzNjQyNmNmZGEyOWY1OWI2ZDdmM2E0ZTMzNTMsMTY5MDQyNTE2ODQ0MA][]

*  PERM\_PRIORITY = 0x1 << INDEX\_PERM\_PRIORITY 8 优先级队列
 *  PERM\_READ = 0x1 << INDEX\_PERM\_READ 4 读权限
 *  PERM\_WRITE = 0x1 << INDEX\_PERM\_WRITE 2 写权限
 *  PERM\_INHERIT = 0x1 << INDEX\_PERM\_INHERIT 1 继承权限

> 0x1表示16进制的1，0x1 << 3 表示将二进制数 0001 向左移动 3 位，结果为 1000，对应十进制数 8。其他的类似

这里最基本的权限只有这三个，也就是读、写、继承。但是是否拥有读写权限是通过如下几个方法进行位运算

java复制代码public static boolean isWriteable(final int perm) {
            return (perm & PERM_WRITE) == PERM_WRITE;
        }
    
        public static boolean isInherited(final int perm) {
            return (perm & PERM_INHERIT) == PERM_INHERIT;
        }
    
        public static boolean isValid(final String perm) {
            return isValid(Integer.parseInt(perm));
        }

然后我们继续源码分析 这里对请求状态码做了区分，分析让queue权限变成可写还是可读 我们先看看perm &= ~PermName.PERM\_WRITE;

我们原先的权限是7(可继承、读、写)，经过运算就变成5(可继承、可读)了

这里我们是执行停写，如果要恢复写入，就执行PermName.PERM\_READ | PermName.PERM\_WRITE;,变为6(可读、可写、不可继承)

所以可以看到这里是先更新了NameServer的路由信息

值得注意的是这里有一个问题，就是实际这里并不会去更新broker的 topics.json配置信息，只是更新了NameServer的内存里面的topicQueueTable，所以也就是如果broker重启或者NameServer重启还是会恢复写入。 这里为什么不去更新broker的topics.json,个人觉得可能是一个bug或者是为了逻辑的简单

## 消费者更新Topic路由信息 ##

至于消费者是如何更新获取到NameServer更新的topicQueueTable，我们可以去随便看看消费者的代码

我们可以看到消费者在启动的时候就启动了一个定时任务 MQClientInstance.this.updateTopicRouteInfoFromNameServer();

![format_png 8][]

![get_code_MTk5M2YyMGUwZmY1ZjExODFhM2JmNzhhNDczOWMxYTcsMTY5MDQyNTE2ODQ0MA][]

这里定时会去Nameserver拉取最新的Topic路由信息，所以我们更新Nameserver的路由信息后这里就会拉去到。更新时间我们可以看看

![format_png 9][]

![get_code_ZTE4MmQ1NWRkZDI0OWI0ZTZlNzBlYTc0M2Q1ODQ5MDAsMTY5MDQyNTE2ODQ0MA][]

默认30s，也就是说我们在停写命令发出后，最长30s内还是会有消息写入

值得注意的是NameServer之间并不会互相通信，所以如果我们是集群模还需要循环遍历所有NameServer执行该命令

[format_png]: https://img-blog.csdnimg.cn/img_convert/a82b6570ab84cf77bd97b2b7361eff43.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/78cf0df2ca12be69784506a9414e1627.webp?x-oss-process=image/format,png
[get_code_ZGZkM2I1NGUwMTYzODRiM2E0ZjQ3NmMwZWEwZTkwMjQsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=ZGZkM2I1NGUwMTYzODRiM2E0ZjQ3NmMwZWEwZTkwMjQsMTY5MDQyNTE2ODQ0MA==
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/86f472ef2f3efffa4e6368c6e6d35c1c.webp?x-oss-process=image/format,png
[get_code_ZDAyNDFmYjhhMGVkMzc2NzQ2NTUyNjU4NWJmNWU4NTksMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=ZDAyNDFmYjhhMGVkMzc2NzQ2NTUyNjU4NWJmNWU4NTksMTY5MDQyNTE2ODQ0MA==
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/c4609eb75b437603d37e8bf8e27ed54d.webp?x-oss-process=image/format,png
[get_code_MzIxNmVlNGJiMzZmN2YwMTExMTdkNWM0YzdiMjM4MzcsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=MzIxNmVlNGJiMzZmN2YwMTExMTdkNWM0YzdiMjM4MzcsMTY5MDQyNTE2ODQ0MA==
[format_png 4]: https://img-blog.csdnimg.cn/img_convert/674ed043d941e449ef1b4e296adcfb94.webp?x-oss-process=image/format,png
[get_code_NDNlMjNkMGZhZjRiMTc5YTA0N2U3MTIyM2I1MmE3MjIsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=NDNlMjNkMGZhZjRiMTc5YTA0N2U3MTIyM2I1MmE3MjIsMTY5MDQyNTE2ODQ0MA==
[format_png 5]: https://img-blog.csdnimg.cn/img_convert/29e4bd32d115e45c86a175c3ef838cd7.webp?x-oss-process=image/format,png
[get_code_MmI1NzUxOGEyY2Y1NDI2NGFjNzNmMDQzMTdjOTA1OWUsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=MmI1NzUxOGEyY2Y1NDI2NGFjNzNmMDQzMTdjOTA1OWUsMTY5MDQyNTE2ODQ0MA==
[format_png 6]: https://img-blog.csdnimg.cn/img_convert/ef54a0e7bbd743acbccdf69c92654d2d.webp?x-oss-process=image/format,png
[get_code_MGM2MzRjOGFlZDg5NmZiYzJjMzBiM2RmY2RhYmEzMTMsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=MGM2MzRjOGFlZDg5NmZiYzJjMzBiM2RmY2RhYmEzMTMsMTY5MDQyNTE2ODQ0MA==
[format_png 7]: https://img-blog.csdnimg.cn/img_convert/badfa80f81409b7f51fabd50da3aaaf7.webp?x-oss-process=image/format,png
[get_code_YWI1MWQzNjQyNmNmZGEyOWY1OWI2ZDdmM2E0ZTMzNTMsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=YWI1MWQzNjQyNmNmZGEyOWY1OWI2ZDdmM2E0ZTMzNTMsMTY5MDQyNTE2ODQ0MA==
[format_png 8]: https://img-blog.csdnimg.cn/img_convert/ca4560ad29bec2ff25f41f58b13ced02.webp?x-oss-process=image/format,png
[get_code_MTk5M2YyMGUwZmY1ZjExODFhM2JmNzhhNDczOWMxYTcsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=MTk5M2YyMGUwZmY1ZjExODFhM2JmNzhhNDczOWMxYTcsMTY5MDQyNTE2ODQ0MA==
[format_png 9]: https://img-blog.csdnimg.cn/img_convert/679910c826f80f1aa4363f44b501ef39.webp?x-oss-process=image/format,png
[get_code_ZTE4MmQ1NWRkZDI0OWI0ZTZlNzBlYTc0M2Q1ODQ5MDAsMTY5MDQyNTE2ODQ0MA]: https://mp.toutiao.com/mp/agw/article_material/open_image/get?code=ZTE4MmQ1NWRkZDI0OWI0ZTZlNzBlYTc0M2Q1ODQ5MDAsMTY5MDQyNTE2ODQ0MA==