结合别人的文章,做RocketMQ的一点原理分析,结合源码(尽量)----未完待续

Broker


与Namesrv的关系

1.从namesrv获取配置信息

 /**
* BrokerConfig类
*
* broker每隔30秒(此时间无法更改)向所有nameserver发送心跳,心跳包含了自身的topic配置信息。
* 这里的“此时间无法更改”是别人的总结,还没搞懂为啥此时间不可更改,明明原版注释中写的是取值范围在10,000到60,000之间
*
* This configurable item defines interval of topics registration of broker to name server. Allowing values are
* between 10, 000 and 60, 000 milliseconds.
*/
private int registerNameServerPeriod = 1000 * 30;

2.namesrv检查broker的心跳

         /**
* NamesrvController类
* Namesrv定时检查Broker心跳
* 每10秒检查一次,时间戳超过2分钟,则认为Broker失效,从brokerLiveTable中移除
*/
this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override
public void run() {
NamesrvController.this.routeInfoManager.scanNotActiveBroker();
}
}, 5, 10, TimeUnit.SECONDS); ... /**
* RouteInfoManager类
* Namesrv查看不活跃|失效的Broker
*/
public void scanNotActiveBroker() {
Iterator<Entry<String, BrokerLiveInfo>> it = this.brokerLiveTable.entrySet().iterator();
while (it.hasNext()) {
Entry<String, BrokerLiveInfo> next = it.next();
long last = next.getValue().getLastUpdateTimestamp();
if ((last + BROKER_CHANNEL_EXPIRED_TIME) < System.currentTimeMillis()) {
RemotingUtil.closeChannel(next.getValue().getChannel());
it.remove();
log.warn("The broker channel expired, {} {}ms", next.getKey(), BROKER_CHANNEL_EXPIRED_TIME);
this.onChannelDestroy(next.getKey(), next.getValue().getChannel());
}
}
}

3.Namesrv检查与Broker的长连接

 /**
* BrokerHousekeepingService类
* 当NameServer和Broker的长连接断掉以后,onChannelDestroy函数会被调用,把这个Broker的信息清理出去。
*/
@Override
public void onChannelClose(String remoteAddr, Channel channel) {
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
} /**
* BrokerHousekeepingService类
* 当NameServer和Broker的长连接断掉以后,onChannelDestroy函数会被调用,把这个Broker的信息清理出去。
*/
@Override
public void onChannelException(String remoteAddr, Channel channel) {
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
} /**
* BrokerHousekeepingService类
* 当NameServer和Broker的长连接断掉以后,onChannelDestroy函数会被调用,把这个Broker的信息清理出去。
*/
@Override
public void onChannelIdle(String remoteAddr, Channel channel) {
this.namesrvController.getRouteInfoManager().onChannelDestroy(remoteAddr, channel);
}

Consumer|Producer


与Namesrv的关系

     /**
* ClientConfig类
*
* 默认情况下,消费者每隔30秒从nameserver获取所有topic的最新队列情况,
* 这意味着某个broker如果宕机,客户端最多要30秒才能感知。
*
* 默认情况下,生产者每隔30秒从nameserver获取所有topic的最新队列情况,
* 这意味着某个broker如果宕机,生产者最多要30秒才能感知,在此期间,发往该broker的消息发送失败。
*
* Pulling topic information interval from the named server
*/
private int pollNameServerInterval = 1000 * 30;

与Broker的关系

1.客户端(生产者|消费者)向Broker发送心跳

     /**
* ClientConfig类
*
* 默认情况下,消费者|生产者 每隔30秒向所有broker发送心跳
*
* Heartbeat interval in microseconds with message broker
*/
private int heartbeatBrokerInterval = 1000 * 30;

2.Broker定期检查客户端心跳检查

     /**
* ClientHousekeepingService类
*
* Broker定时清理不活动的客户端(生产者|消费者)连接
* 每10秒检查一次,时间戳超过2分钟,则关闭客户端(生产者|消费者)连接
*/
public void start() { this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
@Override
public void run() {
try {
ClientHousekeepingService.this.scanExceptionChannel();
} catch (Throwable e) {
log.error("Error occurred when scan not active client channels.", e);
}
}
}, 1000 * 10, 1000 * 10, TimeUnit.MILLISECONDS);
} ... /**
* ClientHousekeepingService类
*
* Broker定时清理不活动的客户端(生产者|消费者)连接
* 每10秒检查一次,时间戳超过2分钟,则关闭客户端(生产者|消费者)连接
*/
private void scanExceptionChannel() {
this.brokerController.getProducerManager().scanNotActiveChannel();
this.brokerController.getConsumerManager().scanNotActiveChannel();
this.brokerController.getFilterServerManager().scanNotActiveChannel();
}

3.Broker检查与客户端的长连接

     /**
* ClientHousekeepingService类
*
* 当客户端(Producer|Consumer)和Broker的长连接断掉以后,doChannelCloseEvent函数会被调用,把对应客户端(Producer|Consumer)断开连接
*/
@Override
public void onChannelClose(String remoteAddr, Channel channel) {
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
} /**
* ClientHousekeepingService类
*
* 当客户端(Producer|Consumer)和Broker的长连接断掉以后,doChannelCloseEvent函数会被调用,把对应客户端(Producer|Consumer)断开连接
*/
@Override
public void onChannelException(String remoteAddr, Channel channel) {
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
} /**
* ClientHousekeepingService类
*
* 当客户端(Producer|Consumer)和Broker的长连接断掉以后,doChannelCloseEvent函数会被调用,把对应客户端(Producer|Consumer)断开连接
*/
@Override
public void onChannelIdle(String remoteAddr, Channel channel) {
this.brokerController.getProducerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getConsumerManager().doChannelCloseEvent(remoteAddr, channel);
this.brokerController.getFilterServerManager().doChannelCloseEvent(remoteAddr, channel);
}
上一篇:使用 Device Mapper来改变Docker容器的大小


下一篇:iOS开发常用快捷键