RocketMQ NameServer

Posted 2021-06-04 levelup

Broker 消息服务器在启动时向所有Name Server 注册，消息生产者（Producer）在发送消息之前先从Name Server 获取Broker 服务器地址列表，然后根据负载算法从列表中选择一台消息服务器进行消息发送。NameServer 与每台Broker 服务器保持长连接，如果检测到Broker宕机， 则从路由注册表中将其移除。但是路由变化不会马上通知消息生产者，为什么要这样设计呢？这是为了降低NameServer 实现的复杂性，在消息发送端提供容错机制来保证消息发送的高可用性，发送失败重试？？？

• 一个Topic 拥有多个消息队列(Kafka中分区的概念)
• 一个Broker 为每一主题默认创建4个读队列4个写队列。
• 多个Broker 组成一个集群，同一个主从的brokerName相同，brokerAddrs中的brokerId 为0 代表Master，大于0 表示Slave 。
• BrokerLivelnfo 中的lastUpdateTimestamp 存储上次收到Broker心跳包的时间。

public class RouteInfoManager {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private final static long BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;
    private final ReadWriteLock lock = new ReentrantReadWriteLock();
    // Topic 消息队列路由信息
    private final HashMap<String, List<QueueData>> topicQueueTable;
    // Broker 基础信息， 包含brokerName 、所属集群名称、主备Broker地址
    private final HashMap<String, BrokerData> brokerAddrTable;
    // Broker 集群信息，存储集群中所有Broker 名称
    private final HashMap<String, Set<String>> clusterAddrTable;
    // Broker 状态信息。NameServer 每次收到心跳包时会替换该信
    private final HashMap<String, BrokerLiveInfo> brokerLiveTable;
    // Broker 上的FilterServer 列表，用于类模式消息过滤
    private final HashMap<String, List<String>> filterServerTable;/
}

class BrokerLiveInfo {
    private long lastUpdateTimestamp;
    private DataVersion dataVersion;
    private Channel channel;
    private String haServerAddr;
}

public class BrokerData implements Comparable<BrokerData> {
    private String cluster;
    private String brokerName;
    private HashMap<Long/* brokerId */, String/* broker address */> brokerAddrs;
}