pmq学习三-mq客户端启动的流程

public class MqClientStartup {
   //spring初始化完成 ,重要
   public static void springInitComplete() {
      MqClient.start();
      monitorConfig();
   }

    //初始化
   public static void init(Environment env1) {
      if (initFlag.compareAndSet(false, true)) {
         env = env1;
         initConfig();
      }
   }

    //监控配置
      private static void monitorConfig() {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                while (isRunning) {
                    updateConfig();
                    Util.sleep(2000);
                }
            }

        });
    }
    //更新配置
    protected static void updateConfig() {
        if (properties == null) {
            properties = MqClient.getContext().getConfig().getProperties();
        }
        setRbTimes();
        setPbTimes();
        setAsynCapacity();
        setMetaMode();
        setPullDeltaTime();
        setPublishAsynTimeout();
    }

 }  

public static boolean start() {
    //启动时，会注册消费者组
   if (startFlag.compareAndSet(false, true)) {
      registerConsumerGroup();
   }
   return false;
}

//可以看到只要启动时才会注册消费者组
public static boolean registerConsumerGroup(Map<String, ConsumerGroupVo> groups) {
        if (groups == null || groups.size() == 0) {
            return false;
        }
         //如果已经初始化，此时执行注册消费组
        if (hasInit()) {
            log.info("已经初始化完成！");
            return doRegisterConsumerGroup(groups);
        } else {
             //否者等待初始化完成，然后执行注册消费组 
            log.warn("系统为初始化，启动异步注册！");
            executor.execute(new Runnable() {
                public void run() {
                    while (!hasInit()) {
                        Util.sleep(2000);
                    }
                doRegisterConsumerGroup(groups);
            });
            return true;
        }
    }    

private static boolean doRegisterConsumerGroup(Map<String, ConsumerGroupVo> groups) {

    //变量消费者组
   for (ConsumerGroupVo consumerGroup : groups.values()) {
     //校验消费组  
      if (!checkVaild(consumerGroup)) {
         return false;
      }
       //如果拿到的消费者组版本包含消费者组拿到的元数据中的名称，则设置为false，同时提示已订阅
      if (mqContext.getConsumerGroupVersion().containsKey(consumerGroup.getMeta().getName())) {
         log.info("ConsumerGroup:" + consumerGroup.getMeta().getName() + " has  subscribed,已订阅！");
         return false;
      }
       //消费者组通过元数据拿到originName如果为空，则设置originName
      if (Util.isEmpty(consumerGroup.getMeta().getOriginName())) {
         consumerGroup.getMeta().setOriginName(consumerGroup.getMeta().getName());
      }
       //消费组拿到topic如果不为空，则消费组名称放入信息
      if (consumerGroup.getTopics() != null) {
         consumerGroupNames.put(consumerGroup.getMeta().getOriginName(),
               new ArrayList<>(consumerGroup.getTopics().keySet()));
      } else {
         consumerGroupNames.put(consumerGroup.getMeta().getOriginName(), new ArrayList<>());
      }
      groupNames += consumerGroup.getMeta().getName() + ",";
   }
   //事件触发,将注册设置为true
   register();
    //消费组注册请求，request设置信息包括：消费者名称、消费者id、客户端id、消费者名称、设置subEnv
   ConsumerGroupRegisterRequest request = new ConsumerGroupRegisterRequest();
   request.setConsumerGroupNames(consumerGroupNames);
   request.setConsumerId(mqContext.getConsumerId());
   request.setClientIp(mqContext.getConfig().getIp());
   request.setConsumerName(mqContext.getConsumerName());
   if (MqClient.getMqEnvironment() != null) {
      if (MqEnv.FAT == MqClient.getMqEnvironment().getEnv()) {
         request.setSubEnv(MqClient.getMqEnvironment().getSubEnv().toLowerCase());
      }
   }
   try {
      //消费者组注册响应，注册消费者，放入请求
      ConsumerGroupRegisterResponse consumerGroupRegisterResponse = mqContext.getMqResource()
            .registerConsumerGroup(request);
       //如果消费者组注册响应为success，则填充信息
      if (consumerGroupRegisterResponse.isSuc()) {
         Map<String, String> broadcastConsumerGroupNames = consumerGroupRegisterResponse
               .getConsumerGroupNameNew();
         for (ConsumerGroupVo consumerGroup : groups.values()) {
            if (broadcastConsumerGroupNames != null
                  && broadcastConsumerGroupNames.containsKey(consumerGroup.getMeta().getOriginName())) {
               consumerGroup.getMeta()
                     .setName(broadcastConsumerGroupNames.get(consumerGroup.getMeta().getOriginName()));
            }
            mqContext.getConfigConsumerGroup().put(consumerGroup.getMeta().getName(), consumerGroup);
            mqContext.getConsumerGroupVersion().put(consumerGroup.getMeta().getName(), 0L);
            //fire消费者组注册事件为true getRegisterConsumerGroupListeners()线程启动
             fireConsumerGroupRegisterEvent(consumerGroup);
         }
          //创建消费者polling服务，并启动consumerPollingService服务，mqCheckService创建，同时启动
         consumerPollingService = mqFactory.createConsumerPollingService();
         consumerPollingService.start();
         mqCheckService = mqFactory.createMqCheckService();
         mqCheckService.start();
         // MqCheckService.getInstance().start(mqContext);
         log.info(groupNames + "  subscribe_suc,订阅成功！and json is " + JsonUtil.toJson(request));
      } else {
         throw new RuntimeException("registerConsumerGroup_error, the req is" + JsonUtil.toJsonNull(request)
               + ",and resp is " + JsonUtil.toJson(consumerGroupRegisterResponse));
      }
   } catch (Exception e) {
      log.error("consumer_group_register_error", e);
      throw new RuntimeException(e);
   }
   return true;
}

@Override
public void start() {
   if (startFlag.compareAndSet(false, true)) {
      isStop = false;
      runStatus = false;
      executor = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(100),
            SoaThreadFactory.create("ConsumerPollingService", true),
            new ThreadPoolExecutor.DiscardOldestPolicy());
      executor.execute(new Runnable() {
         @Override
         public void run() {
            while (!isStop) {
                //将允许状态设置为true，同时将traceMessaageItem设置为true，执行longPolling操作
               TraceMessageItem traceMessageItem = new TraceMessageItem();
               runStatus = true;
               try {
                  traceMessageItem.status = "suc";
                  longPolling();
               } catch (Exception e) {
                  // e.printStackTrace();
                  traceMessageItem.status = "fail";
                  Util.sleep(1000);
               }
                //链路追踪添加链路信息，同时设置为false
               traceMsg.add(traceMessageItem);
               runStatus = false;
            }
         }
      });
   }
}

protected void longPolling() {
   if (mqContext.getConsumerId() > 0 && mqContext.getConsumerGroupVersion() != null
         && mqContext.getConsumerGroupVersion().size() > 0) {
       //设置事务
      Transaction transaction = Tracer.newTransaction("mq-group", "longPolling");
      try {
         //创建获取消费组请求对象，放入消费者id、消费者组版本 
         GetConsumerGroupRequest request = new GetConsumerGroupRequest();
         request.setConsumerId(mqContext.getConsumerId());
         request.setConsumerGroupVersion(mqContext.getConsumerGroupVersion());
          //获取消费者组响应
         GetConsumerGroupResponse response = mqResource.getConsumerGroup(request);
         if (response != null && response.getConsumerDeleted() == 1) {
            log.info("consumerid为" + request.getConsumerId());
         }
          //执行处理group
         handleGroup(response);
         transaction.setStatus(Transaction.SUCCESS);
      } catch (Exception e) {
         transaction.setStatus(e);
      } finally {
         transaction.complete();
      }
   } else {
      Util.sleep(1000);
   }
}

protected void handleGroup(GetConsumerGroupResponse response) {
   if (isStop) {
      return;
   }
    //响应如果不为空，则设置broker元数据模式
   if (response != null) {
      mqContext.setBrokerMetaMode(response.getBrokerMetaMode());
      if (MqClient.getMqEnvironment() != null && MqClient.getMqEnvironment().getEnv() == MqEnv.FAT) {
         MqClient.getContext().setAppSubEnvMap(response.getConsumerGroupSubEnvMap());
      }
   }
    //mq客户端重新启动
   if (response != null && response.getConsumerDeleted() == 1) {
      MqClient.reStart();
      Util.sleep(5000);
      return;
   } else if (response != null && response.getConsumerGroups() != null
         && response.getConsumerGroups().size() > 0) {
      log.info("get_consumer_group_data,获取到的最新消费者组数据为：" + JsonUtil.toJson(response));
      TraceMessageItem item = new TraceMessageItem();
      item.status = "changed";
      item.msg = JsonUtil.toJson(response);
       //响应拿到消费者组，进行遍历，如果没有stop，同时请求为新请求，请求放入，同时创建mq组线程服务
      response.getConsumerGroups().entrySet().forEach(t1 -> {
         if (!isStop) {
            if (!mqExcutors.containsKey(t1.getKey())) {
               mqExcutors.put(t1.getKey(), mqFactory.createMqGroupExcutorService());
            }
            log.info("consumer_group_data_change,消费者组" + t1.getKey() + "发生重平衡或者meta更新");
            // 进行重平衡操作或者更新元数据信息
            mqExcutors.get(t1.getKey()).rbOrUpdate(t1.getValue(), response.getServerIp());  
            mqContext.getConsumerGroupVersion().put(t1.getKey(), t1.getValue().getMeta().getVersion());
         }
      });
      traceMsg.add(item);
   }
   // 然后启动
   mqExcutors.values().forEach(t1 -> {
      t1.start();
   });
}

public MqGroupExcutorService(IMqResource mqResource) {
   //mq上下文、mqResource、mq工厂 
   this.mqContext = MqClient.getContext();
   this.mqResource = mqResource;
   this.mqFactory = MqClient.getMqFactory();
}

public MqCheckService(IMqResource mqResource) {
   this.mqContext = MqClient.getContext();
   this.mqResource = mqResource;
}

@Component
public class MqStartProcessor implements BeanFactoryPostProcessor, PriorityOrdered, EnvironmentAware {
   @Override
   public void postProcessBeanFactory(ConfigurableListableBeanFactory beanFactory) throws BeansException {
      if (environment != null) { 
         if (initFlag.compareAndSet(false, true)) { 
            logger.info("消息客户端开始初始化！");
            MqClient.setSubscriberResolver(new SubscriberResolver());
             //初始化
            MqClientStartup.init(environment);
            //MqClientStartup.start();
            // statService.start();
            logger.info("消息客户端初始化完成！");
         }
      }
   }

}

public static void init(Environment env1) {
   if (initFlag.compareAndSet(false, true)) {
      env = env1;
      initConfig();
   }
}

//初始化配置
private static void initConfig() {
        MqConfig config = new MqConfig();
        String netCard = System.getProperty("mq.network.netCard", env.getProperty("mq.network.netCard", ""));       
        String url =System.getProperty("mq.broker.url", env.getProperty("mq.broker.url", ""));      
        String host = System.getProperty("mq.client.host", env.getProperty("mq.client.host", ""));
        String serverPort = System.getProperty("server.port", env.getProperty("server.port", "8080"));
        String asynCapacity = System.getProperty("mq.asyn.capacity", env.getProperty("mq.asyn.capacity", "2000"));
        String rbTimes = System.getProperty("mq.rb.times", env.getProperty("mq.rb.times", "4"));
        String pbRetryTimes = System.getProperty("mq.pb.retry.times", env.getProperty("mq.pb.retry.times", "10"));
        String readTimeOut = System.getProperty("mq.http.timeout", env.getProperty("mq.http.timeout", "10000"));
        String pullDeltaTime = System.getProperty("mq.pull.time.delta", env.getProperty("mq.pull.time.delta", "150"));
        boolean metaMode = "true"
                .equals(System.getProperty("mq.broker.metaMode", env.getProperty("mq.broker.metaMode", "true")));

         //mq客户端初始化，然后更新配置 
        MqClient.init(config);
        updateConfig();
    }

public static void init(MqConfig config) {
   if (initFlag.compareAndSet(false, true)) {
      //执行初始化 
      doInit(config);
      //激活初始化事件  getInitCompleted 线程启动
      fireInitEvent();
      log.info("mq_client has  inited，初始化完成");
   }
}

//执行初始化
private static void doInit(MqConfig config) {
        //设置消费者名称
        mqContext.setConsumerName(
                ConsumerUtil.getConsumerId(config.getIp(), PropUtil.getProcessId() + "", config.getServerPort()));
         //如果mq上下文里面的mq信息为null，mqContext填充配置信息，同时异步消息为空，则创建msgAsyn队列。
        if (mqContext.getMqResource() == null) {
            mqContext.setMqResource(
                    getMqFactory().createMqResource(config.getUrl(), config.getReadTimeOut(), config.getReadTimeOut()));
        }
        mqContext.setConfig(config);
        if (msgsAsyn == null) {
            msgsAsyn = new ArrayBlockingQueue<>(config.getAsynCapacity());
        }
        //创建mqBrokerUrlRefresh服务，同时启动mqBrokerUrl刷新服务
        mqBrokerUrlRefreshService = mqFactory.createMqBrokerUrlRefreshService();
        mqBrokerUrlRefreshService.start();
    }

@Override
public void start() {
   if (startFlag.compareAndSet(false, true)) {          
      isStop = false;
      runStatus = false;
      //执行更新brokerUrls操作，使用单例线程池
      doUpdateBrokerUrls();
      executor = Executors.newScheduledThreadPool(1,
            SoaThreadFactory.create("mq-brokerFreshService-pool-%d", Thread.MAX_PRIORITY - 1, true));
      executor.scheduleAtFixedRate(new Runnable() {
         @Override
         public void run() {
            if (!isStop) {
               runStatus = true;
               doUpdateBrokerUrls();
               runStatus = false;
            }
         }
      }, 1, 20, TimeUnit.SECONDS);
   }
}

protected void doUpdateBrokerUrls() {
        try {
            GetMetaGroupResponse response = this.mqResource.getMetaGroup(request);
            if(response==null){
                return;
            }
            if (response != null && response.isSuc()) {
                mqContext.setBrokerMetaMode(response.getBrokerMetaMode());
                mqContext.setMetricUrl(response.getMetricUrl());
                if(Util.isEmpty(mqContext.getMetricUrl())){
                    //MqMeticReporterService.getInstance(mqClientBase).close();
                    MqClient.getMqFactory().createMqMeticReporterService().close();
                }else{
                    //创建mqMeticReportService，并启动
                    MqClient.getMqFactory().createMqMeticReporterService().start();
                }
            }           
            if (mqContext.getBrokerMetaMode() == 1 || (mqContext.getBrokerMetaMode() == 0 && mqContext.getConfig().isMetaMode())) {
                //List<String> brokerUrls = response.getBrokerIp();
                if (response.getBrokerIpG1()!=null) {                   
                    mqContext.setBrokerUrls(response.getBrokerIpG1(),response.getBrokerIpG2());
                }
            } else if (mqContext.getBrokerMetaMode() == -1 || !mqContext.getConfig().isMetaMode()) {
                mqContext.setBrokerUrls(new ArrayList<>(),new ArrayList<>());
            }
        } catch (Exception e) {
            log.error("updateBrokerError", e);
        }

    }

@Override
public void start() {
   if (startFlag.compareAndSet(false, true)) {
      // 每30s上报数据
      reporter.start(30, TimeUnit.SECONDS);
   }
}