RocketMQ Bug修复记录

系统管理员 2023-03-01 11:39 26阅读 0赞

### 文章目录 ###

*   *   *  1、Bug详情及解决
         *   *  1.1. Bug 来龙去脉
             *  1.2. 验证这真的是一个BUG
             *   *  1.2.1 BrokerFixedThreadPoolExecutor
                 *  1.2.2 FutureTaskExt
                 *  1.2.3 RegisterBrokerResult
                 *  1.2.4 TestChen2
                 *  1.2.5 ThreadFactoryImpl
             *  1.3. 修复BUG
             *   *  1.3.1 TestChen2
         *  2、总结

### 1、Bug详情及解决 ###

#### 1.1. Bug 来龙去脉 ####

这是`RocketMQ`的`核心流程`里面，`BrokerServer`向`NameServer`发送心跳时的一个BUG。

首先呢？我们先看一段源代码。

public List<RegisterBrokerResult> registerBrokerAll(
            final String clusterName,
            final String brokerAddr,
            final String brokerName,
            final long brokerId,
            final String haServerAddr,
            final TopicConfigSerializeWrapper topicConfigWrapper,
            final List<String> filterServerList,
            final boolean oneway,
            final int timeoutMills,
            final boolean compressed) { 
    
            // 初始化一个List，存放每个NameServer注册结果的
            // 多线程 会有并发问题吧
            final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
            /** * 之所以使用 Vector: * 线程安全的集和容器无非就3个： * 1、Collections.synchronizedList()方式： * 2、Vector()方式：适用于写多的场景 * 3、CopyOnWriteArrayList()方式：适用于读多写少的场景 * * 由于此处的场景只是用到了写，所以我们选择了 Vector()方式 */
            // 获取 NameServer 地址列表
            List<String> nameServerAddressList =
                    this.remotingClient.getNameServerAddressList();
            if (nameServerAddressList != null && nameServerAddressList.size() > 0) { 
    
                // 构建请求头，在请求头里面放很多的信息，比如说 BrokerId 和 BrokerName
                final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
                requestHeader.setBrokerAddr(brokerAddr);
                requestHeader.setBrokerId(brokerId);
                requestHeader.setBrokerName(brokerName);
                requestHeader.setClusterName(clusterName);
                requestHeader.setHaServerAddr(haServerAddr);
                requestHeader.setCompressed(compressed);
    
                // 构建请求体，包含一些配置
                RegisterBrokerBody requestBody = new RegisterBrokerBody();
                requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
                requestBody.setFilterServerList(filterServerList);
                final byte[] body = requestBody.encode(compressed);
                final int bodyCrc32 = UtilAll.crc32(body);
                requestHeader.setBodyCrc32(bodyCrc32);
                // 使用CountDownLatch同步计数器，保证注册完全部的 NameServer之后才往下走，
                // 执行其他逻辑
                final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
                // 遍历NameServer 地址列表，使用线程池去注册
                for (final String namesrvAddr : nameServerAddressList) { 
                    brokerOuterExecutor.execute(new Runnable() { 
                        @Override
                        public void run() { 
                            try { 
                                // 调用 registerBroker 真正执行注册
                                RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
                                if (result != null) { 
                                    registerBrokerResultList.add(result);
                                }
    
                                log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
                            } catch (Exception e) { 
                                log.warn("registerBroker Exception, {}", namesrvAddr, e);
                            } finally { 
                                // 注册完，执行 countDownLatch.countDown(); 同步计数器 -1
                                countDownLatch.countDown();
                            }
                        }
                    });
                }
    
                try { 
                    // 等待所有 NameServer 都注册完，才返回注册结果
                    countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) { 
                }
            }
    
            return registerBrokerResultList;
        }

Bug就是出现在这段代码里面。那到底是什么问题呢？其实就是 多线程并发修改 `ArrayList`可能会出现并发安全问题。首先我们看下面这段代码。

// 初始化一个List，存放每个NameServer注册结果的
    // 多线程 会有并发问题吧
    final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();

他的底层是

@GwtCompatible(serializable = true)
      public static <E> ArrayList<E> newArrayList() { 
        return new ArrayList<E>();
      }

本质上就是一个`ArrayList`。ok下面我们看看他是如何被赋值的。

// 遍历NameServer 地址列表，使用线程池去注册
    for (final String namesrvAddr : nameServerAddressList) { 
        brokerOuterExecutor.execute(new Runnable() { 
            @Override
            public void run() { 
                try { 
                    // 调用 registerBroker 真正执行注册
                    RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
                    if (result != null) { 
                        registerBrokerResultList.add(result);
                    }
                    log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
                } catch (Exception e) { 
                    log.warn("registerBroker Exception, {}", namesrvAddr, e);
                } finally { 
                    // 注册完，执行 countDownLatch.countDown(); 同步计数器 -1
                    countDownLatch.countDown();
                }
            }
        });
    }

上述代码的逻辑很简单，就是遍历`NameServer`地址列表，使用线程池去`NameServer`注册。

所以

if (result != null) { 
                        registerBrokerResultList.add(result);
                    }

并发修改`ArrayList`就会出现线程安全问题。这里我们可以搞一个简单的测试。

#### 1.2. 验证这真的是一个BUG ####

首先我们按照上面的代码新建一个类。代码结构如下所示。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNTEwNTk3_size_16_color_FFFFFF_t_70]

##### 1.2.1 BrokerFixedThreadPoolExecutor #####

package com.example.test.arraylistrocketmq;
    
    import java.util.concurrent.*;
    
    /** * @author Chen * @version 1.0 * @date 2020/7/20 9:29 * @description: */
    
    public class BrokerFixedThreadPoolExecutor extends ThreadPoolExecutor { 
        public BrokerFixedThreadPoolExecutor(final int corePoolSize, final int maximumPoolSize, final long keepAliveTime,
                                             final TimeUnit unit,
                                             final BlockingQueue<Runnable> workQueue) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        }
    
        public BrokerFixedThreadPoolExecutor(final int corePoolSize, final int maximumPoolSize, final long keepAliveTime,
                                             final TimeUnit unit,
                                             final BlockingQueue<Runnable> workQueue, final ThreadFactory threadFactory) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        }
    
        public BrokerFixedThreadPoolExecutor(final int corePoolSize, final int maximumPoolSize, final long keepAliveTime,
                                             final TimeUnit unit,
                                             final BlockingQueue<Runnable> workQueue, final RejectedExecutionHandler handler) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        }
    
        public BrokerFixedThreadPoolExecutor(final int corePoolSize, final int maximumPoolSize, final long keepAliveTime,
                                             final TimeUnit unit,
                                             final BlockingQueue<Runnable> workQueue, final ThreadFactory threadFactory,
                                             final RejectedExecutionHandler handler) { 
            super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        }
    
        @Override
        protected <T> RunnableFuture<T> newTaskFor(final Runnable runnable, final T value) { 
            return new FutureTaskExt<T>(runnable, value);
        }
    }

##### 1.2.2 FutureTaskExt #####

package com.example.test.arraylistrocketmq;
    
    import java.util.concurrent.Callable;
    import java.util.concurrent.FutureTask;
    
    /** * @author Chen * @version 1.0 * @date 2020/7/20 9:30 * @description: */
    
    public class FutureTaskExt<V> extends FutureTask<V> { 
        private final Runnable runnable;
    
        public FutureTaskExt(final Callable<V> callable) { 
            super(callable);
            this.runnable = null;
        }
    
        public FutureTaskExt(final Runnable runnable, final V result) { 
            super(runnable, result);
            this.runnable = runnable;
        }
    
        public Runnable getRunnable() { 
            return runnable;
        }
    }

##### 1.2.3 RegisterBrokerResult #####

package com.example.test.arraylistrocketmq;
    
    import lombok.Data;
    
    /** * @author Chen * @version 1.0 * @date 2020/7/20 9:21 * @description: */
    @Data
    public class RegisterBrokerResult { 
        private String haServerAddr;
        private String masterAddr;
    }

##### 1.2.4 TestChen2 #####

package com.example.test.arraylistrocketmq;
    
    import com.google.common.collect.Lists;
    import org.junit.Test;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /** * @author Chen * @version 1.0 * @date 2020/7/20 9:20 * @description: */
    public class TestChen2 { 
    
    
        @Test
        public void test() throws InterruptedException { 
            for (int i = 0; i < 100; i++) { 
                int finalI = i;
                Thread thread = new Thread(new Runnable() { 
                    @Override
                    public void run() { 
                        deal(finalI);
                    }
                });
                thread.start();
    
                thread.join();
            }
        }
    
        public synchronized void deal(int index) { 
            // 初始化一个List，存放每个NameServer注册结果的
            final List<RegisterBrokerResult> registerBrokerResultList =
                    Lists.newArrayList();
            // 获取 NameServer 地址列表
            List<String> nameServerAddressList = new ArrayList<>();
            for (int j = 0; j < 10; j++) { 
                nameServerAddressList.add("192.168.0." + j);
            }
            AtomicInteger atomicInteger = new AtomicInteger(0);
            // 执行其他逻辑
            final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
    
            BrokerFixedThreadPoolExecutor brokerOuterExecutor = new BrokerFixedThreadPoolExecutor(4, 10, 1, TimeUnit.MINUTES,
                    new ArrayBlockingQueue<Runnable>(32), new ThreadFactoryImpl("brokerOutApi_thread_", true));
    
            // 遍历NameServer 地址列表，使用线程池去注册
            for (final String namesrvAddr : nameServerAddressList) { 
                brokerOuterExecutor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            // 调用 registerBroker 真正执行注册
                            RegisterBrokerResult result = new RegisterBrokerResult();
                            result.setHaServerAddr("chen" + atomicInteger.getAndIncrement());
                            Thread.sleep(100);
                            if (result != null) { 
                                // 注册成功结果放到一个List里去
                                registerBrokerResultList.add(result);
                            }
    
                        } catch (Exception e) { 
                            System.out.println("-----------wei---------------> " + index);
                            System.out.println(e);
                            System.out.println("-----------wei---------------> " + index);
                        } finally { 
                            // 注册完，执行 countDownLatch.countDown(); 同步计数器 -1
                            countDownLatch.countDown();
                        }
                    }
                });
            }
    
            try { 
                // 等待所有 NameServer 都注册完，才返回注册结果
                countDownLatch.await(1000000, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) { 
            }
    
            System.out.println(registerBrokerResultList.size());
            System.out.println("-----------chen---------------------------------------> " + index);
        }
    }

##### 1.2.5 ThreadFactoryImpl #####

package com.example.test.arraylistrocketmq;
    
    import java.util.concurrent.ThreadFactory;
    import java.util.concurrent.atomic.AtomicLong;
    
    /** * @author Chen * @version 1.0 * @date 2020/7/20 9:31 * @description: */
    public class ThreadFactoryImpl implements ThreadFactory { 
        private final AtomicLong threadIndex = new AtomicLong(0);
        private final String threadNamePrefix;
        private final boolean daemon;
    
        public ThreadFactoryImpl(final String threadNamePrefix) { 
            this(threadNamePrefix, false);
        }
    
        public ThreadFactoryImpl(final String threadNamePrefix, boolean daemon) { 
            this.threadNamePrefix = threadNamePrefix;
            this.daemon = daemon;
        }
    
        @Override
        public Thread newThread(Runnable r) { 
            Thread thread = new Thread(r, threadNamePrefix + this.threadIndex.incrementAndGet());
            thread.setDaemon(daemon);
            return thread;
        }
    }

这里我将无关的代码都删除了，只保留了上述代码。我们运行`TestChen2`。

运行结果如下：

com.example.test.arraylistrocketmq.TestChen2
    10
    -----------chen---------------------------------------> 0
    10
    -----------chen---------------------------------------> 1
    10
    -----------chen---------------------------------------> 2
    10
    -----------chen---------------------------------------> 3
    10
    -----------chen---------------------------------------> 4
    10
    -----------chen---------------------------------------> 5
    10
    -----------chen---------------------------------------> 6
    10
    -----------chen---------------------------------------> 7
    10
    -----------chen---------------------------------------> 8
    10
    -----------chen---------------------------------------> 9
    10
    -----------chen---------------------------------------> 10
    9
    -----------chen---------------------------------------> 11
    10
    -----------chen---------------------------------------> 12
    9
    -----------chen---------------------------------------> 13
    10
    -----------chen---------------------------------------> 14
    10
    -----------chen---------------------------------------> 15
    10
    -----------chen---------------------------------------> 16
    10
    -----------chen---------------------------------------> 17
    10
    -----------chen---------------------------------------> 18
    10
    -----------chen---------------------------------------> 19
    10
    -----------chen---------------------------------------> 20
    9
    -----------chen---------------------------------------> 21
    10
    -----------chen---------------------------------------> 22
    10
    -----------chen---------------------------------------> 23
    10
    -----------chen---------------------------------------> 24
    10
    -----------chen---------------------------------------> 25
    10
    -----------chen---------------------------------------> 26
    9
    -----------chen---------------------------------------> 27
    10
    -----------chen---------------------------------------> 28
    10
    -----------chen---------------------------------------> 29
    10
    -----------chen---------------------------------------> 30
    10
    -----------chen---------------------------------------> 31
    10
    -----------chen---------------------------------------> 32
    10
    -----------chen---------------------------------------> 33
    10
    -----------chen---------------------------------------> 34
    10
    -----------chen---------------------------------------> 35
    10
    -----------chen---------------------------------------> 36
    10
    -----------chen---------------------------------------> 37
    10
    -----------chen---------------------------------------> 38
    10
    -----------chen---------------------------------------> 39
    10
    -----------chen---------------------------------------> 40
    10
    -----------chen---------------------------------------> 41
    10
    -----------chen---------------------------------------> 42
    6
    -----------chen---------------------------------------> 43
    10
    -----------chen---------------------------------------> 44
    10
    -----------chen---------------------------------------> 45
    9
    -----------chen---------------------------------------> 46
    8
    -----------chen---------------------------------------> 47
    10
    -----------chen---------------------------------------> 48
    10
    -----------chen---------------------------------------> 49
    10
    -----------chen---------------------------------------> 50
    10
    -----------chen---------------------------------------> 51
    9
    -----------chen---------------------------------------> 52
    10
    -----------chen---------------------------------------> 53
    10
    -----------chen---------------------------------------> 54
    10
    -----------chen---------------------------------------> 55
    10
    -----------chen---------------------------------------> 56
    10
    -----------chen---------------------------------------> 57
    10
    -----------chen---------------------------------------> 58
    10
    -----------chen---------------------------------------> 59
    10
    -----------chen---------------------------------------> 60
    10
    -----------chen---------------------------------------> 61
    10
    -----------chen---------------------------------------> 62
    10
    -----------chen---------------------------------------> 63
    10
    -----------chen---------------------------------------> 64
    10
    -----------chen---------------------------------------> 65
    10
    -----------chen---------------------------------------> 66
    8
    -----------chen---------------------------------------> 67
    10
    -----------chen---------------------------------------> 68
    10
    -----------chen---------------------------------------> 69
    10
    -----------chen---------------------------------------> 70
    8
    -----------chen---------------------------------------> 71
    10
    -----------chen---------------------------------------> 72
    10
    -----------chen---------------------------------------> 73
    10
    -----------chen---------------------------------------> 74
    10
    -----------chen---------------------------------------> 75
    10
    -----------chen---------------------------------------> 76
    10
    -----------chen---------------------------------------> 77
    10
    -----------chen---------------------------------------> 78
    10
    -----------chen---------------------------------------> 79
    10
    -----------chen---------------------------------------> 80
    10
    -----------chen---------------------------------------> 81
    10
    -----------chen---------------------------------------> 82
    10
    -----------chen---------------------------------------> 83
    10
    -----------chen---------------------------------------> 84
    10
    -----------chen---------------------------------------> 85
    9
    -----------chen---------------------------------------> 86
    10
    -----------chen---------------------------------------> 87
    10
    -----------chen---------------------------------------> 88
    10
    -----------chen---------------------------------------> 89
    10
    -----------chen---------------------------------------> 90
    9
    -----------chen---------------------------------------> 91
    10
    -----------chen---------------------------------------> 92
    10
    -----------chen---------------------------------------> 93
    10
    -----------chen---------------------------------------> 94
    10
    -----------chen---------------------------------------> 95
    10
    -----------chen---------------------------------------> 96
    9
    -----------chen---------------------------------------> 97
    9
    -----------chen---------------------------------------> 98
    10
    -----------chen---------------------------------------> 99
    
    Process finished with exit code 0

我们发现并不是每次 `registerBrokerResultList`的大小都是 10，也就是出现了并发安全问题。这里之所以在测试的时候在`deal`方法前面加上`synchronized`，是因为调用`registerBrokerAll`方法的函数是`synchronized`。但是`public synchronized void registerBrokerAll()` 这段代码只是让下面这段代码

// 启动了一个定时调度的任务，让他去给 NameServer 进行注册
            // 默认 30s 发送一次
            this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { 
    
                @Override
                public void run() { 
                    try { 
                        // 将自己注册到NameServer
                        BrokerController.this.
                                registerBrokerAll(true, false, brokerConfig.isForceRegister());
                    } catch (Throwable e) { 
                        log.error("registerBrokerAll Exception", e);
                    }
                }
            }, 1000 * 10, Math.max(10000, Math.min(brokerConfig.getRegisterNameServerPeriod(), 60000)), TimeUnit.MILLISECONDS);

线程池里面的线程安全。也就是说 ，他只是保证了定时任务不同批次任务的线程安全。但是仍然无法保证定时任务同一个批次内的线程安全。比如说：某次发送心跳。Broker需要向 10个NameServer 发送心跳。`public List<RegisterBrokerResult> registerBrokerAll`方法是开启10个线程去发送心跳，如果此时同时返回结果，同时修改`registerBrokerResultList`这个`ArrayList`仍然是有并发安全问题的。

`public synchronized void registerBrokerAll()` 这段代码只是保证 比如说 时刻 0 和时刻 30 。他们发送的心跳没有并发安全问题。

#### 1.3. 修复BUG ####

其实解决并发安全问题我们也是有几种解决方案的。比如用线程安全的集和容器，而线程安全的集和容器无非就3个：

1、Collections.synchronizedList()方式：
    2、Vector()方式：适用于写多的场景
    3、CopyOnWriteArrayList()方式：适用于读多写少的场景

但是直接使用上述方案，锁太粗，不利于高并发，所以我们只要在`list.add()`方法上加个锁就行了。即修改`TestChen2`代码如下：

##### 1.3.1 TestChen2 #####

package com.example.test.arraylistrocketmq;
    
    import com.google.common.collect.Lists;
    import org.junit.Test;
    
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.CountDownLatch;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /** * @author Chen * @version 1.0 * @date 2020/7/20 9:20 * @description: */
    public class TestChen2 { 
    
    
        @Test
        public void test() throws InterruptedException { 
            for (int i = 0; i < 100; i++) { 
                int finalI = i;
                Thread thread = new Thread(new Runnable() { 
                    @Override
                    public void run() { 
                        deal(finalI);
                    }
                });
                thread.start();
    
                thread.join();
            }
        }
    
        public synchronized void deal(int index) { 
            // 初始化一个List，存放每个NameServer注册结果的
            final List<RegisterBrokerResult> registerBrokerResultList =
                    Lists.newArrayList();
            // 获取 NameServer 地址列表
            List<String> nameServerAddressList = new ArrayList<>();
            for (int j = 0; j < 10; j++) { 
                nameServerAddressList.add("192.168.0." + j);
            }
            AtomicInteger atomicInteger = new AtomicInteger(0);
            // 执行其他逻辑
            final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
    
            BrokerFixedThreadPoolExecutor brokerOuterExecutor = new BrokerFixedThreadPoolExecutor(4, 10, 1, TimeUnit.MINUTES,
                    new ArrayBlockingQueue<Runnable>(32), new ThreadFactoryImpl("brokerOutApi_thread_", true));
    
            // 遍历NameServer 地址列表，使用线程池去注册
            for (final String namesrvAddr : nameServerAddressList) { 
                brokerOuterExecutor.execute(new Runnable() { 
                    @Override
                    public void run() { 
                        try { 
                            // 调用 registerBroker 真正执行注册
                            RegisterBrokerResult result = new RegisterBrokerResult();
                            result.setHaServerAddr("chen" + atomicInteger.getAndIncrement());
                            Thread.sleep(10);
                            if (result != null) { 
                                // 注册成功结果放到一个List里去
                                // 锁细化
                                // make ArrayList thread safe, here we don't need to
                                // use Vector or Collections.synchronizedList() or CopyOnWriteArrayList
                                synchronized (registerBrokerResultList) { 
                                    registerBrokerResultList.add(result);
                                }
                            }
    
                        } catch (Exception e) { 
                            System.out.println("-----------wei---------------> " + index);
                            System.out.println(e);
                            System.out.println("-----------wei---------------> " + index);
                        } finally { 
                            // 注册完，执行 countDownLatch.countDown(); 同步计数器 -1
                            countDownLatch.countDown();
                        }
                    }
                });
            }
    
            try { 
                // 等待所有 NameServer 都注册完，才返回注册结果
                countDownLatch.await(1000000, TimeUnit.MILLISECONDS);
            } catch (InterruptedException e) { 
            }
    
            System.out.println(registerBrokerResultList.size());
            System.out.println("-----------chen---------------------------------------> " + index);
        }
    }

核心的就是修改：

if (result != null) { 
        // 注册成功结果放到一个List里去
        // 锁细化
        // make ArrayList thread safe, here we don't need to
        // use Vector or Collections.synchronizedList() or CopyOnWriteArrayList
        synchronized (registerBrokerResultList) { 
            registerBrokerResultList.add(result);
        }
    }

ok ,保存之后再运行一遍。运行结果如下：

10
    -----------chen---------------------------------------> 0
    10
    -----------chen---------------------------------------> 1
    10
    -----------chen---------------------------------------> 2
    10
    -----------chen---------------------------------------> 3
    10
    -----------chen---------------------------------------> 4
    10
    -----------chen---------------------------------------> 5
    10
    -----------chen---------------------------------------> 6
    10
    -----------chen---------------------------------------> 7
    10
    -----------chen---------------------------------------> 8
    10
    -----------chen---------------------------------------> 9
    10
    -----------chen---------------------------------------> 10
    10
    -----------chen---------------------------------------> 11
    10
    -----------chen---------------------------------------> 12
    10
    -----------chen---------------------------------------> 13
    10
    -----------chen---------------------------------------> 14
    10
    -----------chen---------------------------------------> 15
    10
    -----------chen---------------------------------------> 16
    10
    -----------chen---------------------------------------> 17
    10
    -----------chen---------------------------------------> 18
    10
    -----------chen---------------------------------------> 19
    10
    -----------chen---------------------------------------> 20
    10
    -----------chen---------------------------------------> 21
    10
    -----------chen---------------------------------------> 22
    10
    -----------chen---------------------------------------> 23
    10
    -----------chen---------------------------------------> 24
    10
    -----------chen---------------------------------------> 25
    10
    -----------chen---------------------------------------> 26
    10
    -----------chen---------------------------------------> 27
    10
    -----------chen---------------------------------------> 28
    10
    -----------chen---------------------------------------> 29
    10
    -----------chen---------------------------------------> 30
    10
    -----------chen---------------------------------------> 31
    10
    -----------chen---------------------------------------> 32
    10
    -----------chen---------------------------------------> 33
    10
    -----------chen---------------------------------------> 34
    10
    -----------chen---------------------------------------> 35
    10
    -----------chen---------------------------------------> 36
    10
    -----------chen---------------------------------------> 37
    10
    -----------chen---------------------------------------> 38
    10
    -----------chen---------------------------------------> 39
    10
    -----------chen---------------------------------------> 40
    10
    -----------chen---------------------------------------> 41
    10
    -----------chen---------------------------------------> 42
    10
    -----------chen---------------------------------------> 43
    10
    -----------chen---------------------------------------> 44
    10
    -----------chen---------------------------------------> 45
    10
    -----------chen---------------------------------------> 46
    10
    -----------chen---------------------------------------> 47
    10
    -----------chen---------------------------------------> 48
    10
    -----------chen---------------------------------------> 49
    10
    -----------chen---------------------------------------> 50
    10
    -----------chen---------------------------------------> 51
    10
    -----------chen---------------------------------------> 52
    10
    -----------chen---------------------------------------> 53
    10
    -----------chen---------------------------------------> 54
    10
    -----------chen---------------------------------------> 55
    10
    -----------chen---------------------------------------> 56
    10
    -----------chen---------------------------------------> 57
    10
    -----------chen---------------------------------------> 58
    10
    -----------chen---------------------------------------> 59
    10
    -----------chen---------------------------------------> 60
    10
    -----------chen---------------------------------------> 61
    10
    -----------chen---------------------------------------> 62
    10
    -----------chen---------------------------------------> 63
    10
    -----------chen---------------------------------------> 64
    10
    -----------chen---------------------------------------> 65
    10
    -----------chen---------------------------------------> 66
    10
    -----------chen---------------------------------------> 67
    10
    -----------chen---------------------------------------> 68
    10
    -----------chen---------------------------------------> 69
    10
    -----------chen---------------------------------------> 70
    10
    -----------chen---------------------------------------> 71
    10
    -----------chen---------------------------------------> 72
    10
    -----------chen---------------------------------------> 73
    10
    -----------chen---------------------------------------> 74
    10
    -----------chen---------------------------------------> 75
    10
    -----------chen---------------------------------------> 76
    10
    -----------chen---------------------------------------> 77
    10
    -----------chen---------------------------------------> 78
    10
    -----------chen---------------------------------------> 79
    10
    -----------chen---------------------------------------> 80
    10
    -----------chen---------------------------------------> 81
    10
    -----------chen---------------------------------------> 82
    10
    -----------chen---------------------------------------> 83
    10
    -----------chen---------------------------------------> 84
    10
    -----------chen---------------------------------------> 85
    10
    -----------chen---------------------------------------> 86
    10
    -----------chen---------------------------------------> 87
    10
    -----------chen---------------------------------------> 88
    10
    -----------chen---------------------------------------> 89
    10
    -----------chen---------------------------------------> 90
    10
    -----------chen---------------------------------------> 91
    10
    -----------chen---------------------------------------> 92
    10
    -----------chen---------------------------------------> 93
    10
    -----------chen---------------------------------------> 94
    10
    -----------chen---------------------------------------> 95
    10
    -----------chen---------------------------------------> 96
    10
    -----------chen---------------------------------------> 97
    10
    -----------chen---------------------------------------> 98
    10
    -----------chen---------------------------------------> 99
    
    Process finished with exit code 0

从结果我们发现我们的方法是线程安全的。ok,我们修改RocketMQ中的源码

public List<RegisterBrokerResult> registerBrokerAll(
            final String clusterName,
            final String brokerAddr,
            final String brokerName,
            final long brokerId,
            final String haServerAddr,
            final TopicConfigSerializeWrapper topicConfigWrapper,
            final List<String> filterServerList,
            final boolean oneway,
            final int timeoutMills,
            final boolean compressed) { 
    
            // 初始化一个List，存放每个NameServer注册结果的
            // 多线程 会有并发问题吧
            final List<RegisterBrokerResult> registerBrokerResultList = Lists.newArrayList();
            /** * 之所以使用 Vector: * 线程安全的集和容器无非就3个： * 1、Collections.synchronizedList()方式： * 2、Vector()方式：适用于写多的场景 * 3、CopyOnWriteArrayList()方式：适用于读多写少的场景 * * 由于此处的场景只是用到了写，所以我们选择了 Vector()方式 */
            // 获取 NameServer 地址列表
            List<String> nameServerAddressList =
                    this.remotingClient.getNameServerAddressList();
            if (nameServerAddressList != null && nameServerAddressList.size() > 0) { 
    
                // 构建请求头，在请求头里面放很多的信息，比如说 BrokerId 和 BrokerName
                final RegisterBrokerRequestHeader requestHeader = new RegisterBrokerRequestHeader();
                requestHeader.setBrokerAddr(brokerAddr);
                requestHeader.setBrokerId(brokerId);
                requestHeader.setBrokerName(brokerName);
                requestHeader.setClusterName(clusterName);
                requestHeader.setHaServerAddr(haServerAddr);
                requestHeader.setCompressed(compressed);
    
                // 构建请求体，包含一些配置
                RegisterBrokerBody requestBody = new RegisterBrokerBody();
                requestBody.setTopicConfigSerializeWrapper(topicConfigWrapper);
                requestBody.setFilterServerList(filterServerList);
                final byte[] body = requestBody.encode(compressed);
                final int bodyCrc32 = UtilAll.crc32(body);
                requestHeader.setBodyCrc32(bodyCrc32);
                // 使用CountDownLatch同步计数器，保证注册完全部的 NameServer之后才往下走，
                // 执行其他逻辑
                final CountDownLatch countDownLatch = new CountDownLatch(nameServerAddressList.size());
                // 遍历NameServer 地址列表，使用线程池去注册
                for (final String namesrvAddr : nameServerAddressList) { 
                    brokerOuterExecutor.execute(new Runnable() { 
                        @Override
                        public void run() { 
                            try { 
                                // 调用 registerBroker 真正执行注册
                                RegisterBrokerResult result = registerBroker(namesrvAddr,oneway, timeoutMills,requestHeader,body);
                                if (result != null) { 
                                    // 注册成功结果放到一个List里去
                                    // 锁细化
                                    // make ArrayList thread safe, here we don't need to
                                    // use Vector or Collections.synchronizedList() or CopyOnWriteArrayList
                                    synchronized (registerBrokerResultList) { 
                                        registerBrokerResultList.add(result);
                                    }
    
                                }
    
                                log.info("register broker[{}]to name server {} OK", brokerId, namesrvAddr);
                            } catch (Exception e) { 
                                log.warn("registerBroker Exception, {}", namesrvAddr, e);
                            } finally { 
                                // 注册完，执行 countDownLatch.countDown(); 同步计数器 -1
                                countDownLatch.countDown();
                            }
                        }
                    });
                }
    
                try { 
                    // 等待所有 NameServer 都注册完，才返回注册结果
                    countDownLatch.await(timeoutMills, TimeUnit.MILLISECONDS);
                } catch (InterruptedException e) { 
                }
            }
    
            return registerBrokerResultList;
        }

主要就是修改：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNTEwNTk3_size_16_color_FFFFFF_t_70 1]

这里是我在`GitHub`上提交的`Issues` [https://github.com/apache/rocketmq/issues/2170][https_github.com_apache_rocketmq_issues_2170]

### 2、总结 ###

开源的代码直接运行绝大部分是对的，但是总有一些软件的更新使得作者无能为力。之前的API是对的，但是之后就废弃了或修改了是常有的事。所以我们需要跟踪源代码。这只是一个小小的问题，如果没有前辈的无私奉献，很难想象我们自己一天能学到多少内容。感谢各位前辈的辛勤付出，让我们少走了很多的弯路！

点个赞再走呗！欢迎留言哦！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNTEwNTk3_size_16_color_FFFFFF_t_70]: /images/20230208/dcfa0313e6244952bcfb64171ce3a6e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMyNTEwNTk3_size_16_color_FFFFFF_t_70 1]: /images/20230208/c5f3a261b5244ea68894d17e95d4eb4f.png
[https_github.com_apache_rocketmq_issues_2170]: https://github.com/apache/rocketmq/issues/2170