vlambda博客
学习文章列表

    public static void main(String[] args) throws MQClientException, RemotingException, InterruptedException, MQBrokerException { DefaultMQProducer defaultMQProducer = new DefaultMQProducer(GROUP_NAME,false); defaultMQProducer.setNamesrvAddr(NAME_SERVER); defaultMQProducer.setSendMsgTimeout(SEND_TIME_OUT); defaultMQProducer.setVipChannelEnabled(false); //启动的核心方法在这里 defaultMQProducer.start(); System.out.println("==============请选择需要发送堆消息=============="); while (true){ Scanner scanner = new Scanner(System.in); String nextLine = scanner.nextLine(); if("A".equals(nextLine)){ sendTopicA(); } } }

 @Override public void start() throws MQClientException { this.setProducerGroup(withNamespace(this.producerGroup));        //这里是核心的方法区，深入到其中 this.defaultMQProducerImpl.start(); if (null != traceDispatcher) {            try { traceDispatcher.start(this.getNamesrvAddr(), this.getAccessChannel()); } catch (MQClientException e) { log.warn("trace dispatcher start failed ", e); } } }

 public void start(final boolean startFactory) throws MQClientException { switch (this.serviceState) { case CREATE_JUST:                //首先这里会将状态先设置为启动失败，等到服务完全启动之后才修改为启动成功 this.serviceState = ServiceState.START_FAILED;                //一些配置的检查，例如groupName的一些长度限制，是否有非法字符等 this.checkConfig();  if (!this.defaultMQProducer.getProducerGroup().equals(MixAll.CLIENT_INNER_PRODUCER_GROUP)) { this.defaultMQProducer.changeInstanceNameToPID(); }                //创建一个MQ的客户端工厂，主要会和nameserver打交道，                this.mQClientFactory = MQClientManager.getInstance().getOrCreateMQClientInstance(this.defaultMQProducer, rpcHook);
 boolean registerOK = mQClientFactory.registerProducer(this.defaultMQProducer.getProducerGroup(), this); if (!registerOK) { this.serviceState = ServiceState.CREATE_JUST; throw new MQClientException("The producer group[" + this.defaultMQProducer.getProducerGroup() + "] has been created before, specify another name please." + FAQUrl.suggestTodo(FAQUrl.GROUP_NAME_DUPLICATE_URL), null); }                //将一些topic信息存放到这个map集合中 this.topicPublishInfoTable.put(this.defaultMQProducer.getCreateTopicKey(), new TopicPublishInfo());
 if (startFactory) { //真正启动客户端类 mQClientFactory.start(); }
 log.info("the producer [{}] start OK. sendMessageWithVIPChannel={}", this.defaultMQProducer.getProducerGroup(), this.defaultMQProducer.isSendMessageWithVIPChannel());                //将之前的状态同步修改 this.serviceState = ServiceState.RUNNING; break; case RUNNING: case START_FAILED: case SHUTDOWN_ALREADY: throw new MQClientException("The producer service state not OK, maybe started once, " + this.serviceState + FAQUrl.suggestTodo(FAQUrl.CLIENT_SERVICE_NOT_OK), null); default: break; }
 this.mQClientFactory.sendHeartbeatToAllBrokerWithLock();
 this.timer.scheduleAtFixedRate(new TimerTask() { @Override public void run() { try { RequestFutureTable.scanExpiredRequest(); } catch (Throwable e) { log.error("scan RequestFutureTable exception", e); } } }, 1000 * 3, 1000); }

public void start() throws MQClientException {
 synchronized (this) { switch (this.serviceState) { case CREATE_JUST: //又是先声明当前的服务状态为启动失败 this.serviceState = ServiceState.START_FAILED; // If not specified,looking address from name server                    if (null == this.clientConfig.getNamesrvAddr()) { this.mQClientAPIImpl.fetchNameServerAddr(); } // Start request-response channel                    //这里是启动netty的应用，让本地的服务和远程的netty server建立一个channel通道 this.mQClientAPIImpl.start(); // Start various schedule tasks                    //启动了一个定时任务，内部包含了许多额外的定时任务 this.startScheduledTask(); // Start pull service                    //启动拉取消息的后台线程 this.pullMessageService.start(); // Start rebalance service // 启动Rebalance负载均衡服务 this.rebalanceService.start(); // Start push service                    //这里又一次触发了start函数，但是入参事false，所以不会重复执行启动核心部分，但是看不懂为什么要这么设计 this.defaultMQProducer.getDefaultMQProducerImpl().start(false); log.info("the client factory [{}] start OK", this.clientId); //修改状态为已经启动                    this.serviceState = ServiceState.RUNNING; break; case START_FAILED: throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null); default: break; } } }