RocketMQ之消费者启动与消费流程

vivo 互联网服务器团队 - Li Kui

  

• Producer：负责生产消息，一般由业务系统生产消息，可通过集群方式部署。RocketMQ提供多种发送方式，同步发送、异步发送、顺序发送、单向发送。同步和异步方式均需要Broker返回确认信息，单向发送不需要。  
  
• Consumer：负责消费消息，一般是后台系统负责异步消费，可通过集群方式部署。一个消息消费者会从Broker服务器拉取消息、并将其提供给应用程序。提供pull/push两者消费模式。  
  
• Broker Server：负责存储消息、转发消息。RocketMQ系统中负责接收从生产者发送来的消息并存储、同时为消费者的拉取请求作准备，存储消息相关的元数据，包括消费者组、消费进度偏移和主题和队列消息等。  
  
• Name Server：名字服务，充当路由消息的提供者。生产者或消费者能够通过名字服务查找各主题相应的Broker IP列表。多个NameServer实例组成集群，相互独立，没有信息交换。
  

public class Consumer {
  public static void main(String[] args) throws InterruptedException, MQClientException {
    // 实例化消费者
    DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TestConsumer");
    // 设置NameServer的地址
    consumer.setNamesrvAddr("localhost:9876");
    // 订阅一个Topic，以及Tag来过滤需要消费的消息
    consumer.subscribe("Test", "*");
    // 注册回调实现类来处理从broker拉取回来的消息
    consumer.registerMessageListener(new MessageListenerConcurrently() {
      @Override
      public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
        // 标记该消息已经被成功消费
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
      }
    });
    // 启动消费者实例
    consumer.start();
    System.out.printf("Consumer Started.%n");
  }
}

// 实例化消费者
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("TestConsumer");

public DefaultMQPushConsumer(final String consumerGroup, RPCHook rpcHook,
     AllocateMessageQueueStrategy allocateMessageQueueStrategy) {
     this.consumerGroup = consumerGroup;
     this.allocateMessageQueueStrategy = allocateMessageQueueStrategy;
     defaultMQPushConsumerImpl = new DefaultMQPushConsumerImpl(this, rpcHook);
   }

// 设置NameServer的地址
consumer.setNamesrvAddr("localhost:9876");
// 订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
consumer.subscribe("Test", "*");

public static SubscriptionData buildSubscriptionData(final String consumerGroup, String topic,
  String subString) throws Exception {
  SubscriptionData subscriptionData = new SubscriptionData();
  subscriptionData.setTopic(topic);
  subscriptionData.setSubString(subString);
 
  if (null == subString || subString.equals(SubscriptionData.SUB_ALL) || subString.length() == 0) {
    subscriptionData.setSubString(SubscriptionData.SUB_ALL);
  } else {
    String[] tags = subString.split("\\|\\|");
    if (tags.length > 0) {
      for (String tag : tags) {
        if (tag.length() > 0) {
          String trimString = tag.trim();
          if (trimString.length() > 0) {
            subscriptionData.getTagsSet().add(trimString);
            subscriptionData.getCodeSet().add(trimString.hashCode());
          }
        }
      }
    } else {
      throw new Exception("subString split error");
    }
  }
 
  return subscriptionData;
}

public void sendHeartbeatToAllBrokerWithLock() {
  if (this.lockHeartbeat.tryLock()) {
    try {
      this.sendHeartbeatToAllBroker();
      this.uploadFilterClassSource();
    } catch (final Exception e) {
      log.error("sendHeartbeatToAllBroker exception", e);
    } finally {
      this.lockHeartbeat.unlock();
    }
  } else {
    log.warn("lock heartBeat, but failed.");
  }
}

this.mQClientAPIImpl.sendHearbeat(addr, heartbeatData, 3000);

private void uploadFilterClassSource() {
  Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
  while (it.hasNext()) {
    Entry<String, MQConsumerInner> next = it.next();
    MQConsumerInner consumer = next.getValue();
    if (ConsumeType.CONSUME_PASSIVELY == consumer.consumeType()) {
      Set<SubscriptionData> subscriptions = consumer.subscriptions();
      for (SubscriptionData sub : subscriptions) {
        if (sub.isClassFilterMode() && sub.getFilterClassSource() != null) {
          final String consumerGroup = consumer.groupName();
          final String className = sub.getSubString();
          final String topic = sub.getTopic();
          final String filterClassSource = sub.getFilterClassSource();
          try {
            this.uploadFilterClassToAllFilterServer(consumerGroup, className, topic, filterClassSource);
          } catch (Exception e) {
            log.error("uploadFilterClassToAllFilterServer Exception", e);
          }
        }
      }
    }
  }
}

switch (this.defaultMQPushConsumer.getMessageModel()) {
  case BROADCASTING:
    this.offsetStore = new LocalFileOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
    break;
  case CLUSTERING:
    this.offsetStore = new RemoteBrokerOffsetStore(this.mQClientFactory, this.defaultMQPushConsumer.getConsumerGroup());
    break;
  default:
    break;
}

public void start() {
    if (MessageModel.CLUSTERING.equals(ConsumeMessageOrderlyService.this.defaultMQPushConsumerImpl.messageModel())) {
      this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
        @Override
        public void run() {
          ConsumeMessageOrderlyService.this.lockMQPeriodically();
        }
      }, 1000 * 1, ProcessQueue.REBALANCE_LOCK_INTERVAL, TimeUnit.MILLISECONDS);
    }
  }

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
    this.mQClientAPIImpl.start();
    // Start various schedule tasks
    this.startScheduledTask();
    // Start pull service
    this.pullMessageService.start();
    // Start rebalance service
    this.rebalanceService.start();
    // Start push service
    this.defaultMQProducer.getDefaultMQProducerImpl().start(false);
    log.info("the client factory [{}] start OK", this.clientId);
    this.serviceState = ServiceState.RUNNING;
    break;
  case RUNNING:
    break;
  case SHUTDOWN_ALREADY:
    break;
  case START_FAILED:
    throw new MQClientException("The Factory object[" + this.getClientId() + "] has been created before, and failed.", null);
  default:
    break;
}

DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("TestConsumer");
// 设置NameServer的地址
consumer.setNamesrvAddr("localhost:9876");
// 启动消费者实例
consumer.start();
//获取主题下所有的消息队列，这里根据主题从nameserver获取的
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("Test");
for (MessageQueue queue : mqs) {
  //获取当前队列的消费位移，指定消费进度offset，fromstore：从broker中获取还是本地获取，true-broker
  long offset = consumer.fetchConsumeOffset(queue, true);
  PullResult pullResult = null;
  while (offset < pullResult.getMaxOffset()) {
    //第二个参数为tag，获取指定topic下的tag
    //第三个参数表示从哪个位移下开始消费消息
    //第四个参数表示一次最大拉取多少个消息
    try {
      pullResult = consumer.pullBlockIfNotFound(queue, "*", offset, 32);
    } catch (Exception e) {
      e.printStackTrace();
      System.out.println("pull拉取消息失败");
    }
    //代码省略，记录消息位移
    offset = pullResult.getNextBeginOffset();
    //代码省略，这里为消费消息
  }
}

case CREATE_JUST:
         //......
        // Start pull service
        this.pullMessageService.start();
        // Start rebalance service
        this.rebalanceService.start();
        //.......
        this.serviceState = ServiceState.RUNNING;
        break;
  case RUNNING:

public class RebalanceService extends ServiceThread {
  //初始化，省略....
 
  @Override
  public void run() {
    log.info(this.getServiceName() + " service started");
 
    while (!this.isStopped()) {
      this.waitForRunning(waitInterval);
      //做负载均衡
      this.mqClientFactory.doRebalance();
    }
 
    log.info(this.getServiceName() + " service end");
  }
 
  @Override
  public String getServiceName() {
    return RebalanceService.class.getSimpleName();
  }
}

public void doRebalance(final boolean isOrder) {
    Map<String, SubscriptionData> subTable = this.getSubscriptionInner();
    if (subTable != null) {
      for (final Map.Entry<String, SubscriptionData> entry : subTable.entrySet()) {
        final String topic = entry.getKey();
        try {
          /根据 /每个topic，以及它是否顺序消息模式来做rebalance
          this.rebalanceByTopic(topic, isOrder);
        } catch (Throwable e) {
          if (!topic.startsWith(MixAll.RETRY_GROUP_TOPIC_PREFIX)) {
            log.warn("rebalanceByTopic Exception", e);
          }
        }
      }
    }
 
    this.truncateMessageQueueNotMyTopic();
  }

@Override
 public void dispatchPullRequest(List<PullRequest> pullRequestList) {
   for (PullRequest pullRequest : pullRequestList) {
     this.defaultMQPushConsumerImpl.executePullRequestImmediately(pullRequest);
     log.info("doRebalance, {}, add a new pull request {}", consumerGroup, pullRequest);
   }
 }

@Override
  public void run() {
    log.info(this.getServiceName() + " service started");
 
    while (!this.isStopped()) {
      try {
        PullRequest pullRequest = this.pullRequestQueue.take();
        if (pullRequest != null) {
          this.pullMessage(pullRequest);
        }
      } catch (InterruptedException e) {
      } catch (Exception e) {
        log.error("Pull Message Service Run Method exception", e);
      }
    }
 
    log.info(this.getServiceName() + " service end");
  }

private void pullMessage(final PullRequest pullRequest) {
   final MQConsumerInner consumer = this.mQClientFactory.selectConsumer(pullRequest.getConsumerGroup());
   if (consumer != null) {
     DefaultMQPushConsumerImpl impl = (DefaultMQPushConsumerImpl) consumer;
     impl.pullMessage(pullRequest);
   } else {
     log.warn("No matched consumer for the PullRequest {}, drop it", pullRequest);
   }
 }

//Pull回调
PullCallback pullCallback = new PullCallback() {
      @Override
      public void onSuccess(PullResult pullResult) {
        if (pullResult != null) {
          pullResult = DefaultMQPushConsumerImpl.this.pullAPIWrapper.processPullResult(pullRequest.getMessageQueue(), pullResult,
            subscriptionData);
 
          switch (pullResult.getPullStatus()) {
            case FOUND:
             //省略...消费位移更新
                DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
                  pullResult.getMsgFoundList(),
                  processQueue,
                  pullRequest.getMessageQueue(),
                  dispathToConsume);

//监听消息
 status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);

// 注册回调实现类来处理从broker拉取回来的消息
 consumer.registerMessageListener(new MessageListenerConcurrently() {
   @Override
   public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
     System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
     // 标记该消息已经被成功消费
     return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
   }
 });

两者底层实际一样，push模式也是基于pull模式来实现的。  
   pull模式需要我们通过程序主动通过consumer向broker拉消息，而消息的push模式则只需要我们提供一个listener监听，实时获取消息。

push模式采用长轮询阻塞的方式获取消息，实时性非常高； 
 push模式rocketMQ处理了获取消息的细节，使用起来比较简单方便； 
 pull模式可以指定消费进度，想消费多少就消费多少，灵活性大。

push模式当消费者能力远远低于生产者能力的时候，会产生一定的消费者消息堆积；  
   pull模式实时性很低，频率不好设置；  
   拉取消息的间隔不好设置，太短则产生很多无效Pull请求的RPC开销，影响MQ整体的网络性能，太长则实时性差。

对于服务端生产消息数据比较大时，而消费端处理比较复杂，消费能力相对较低时，这种情况就适用pull模式；  
   对于数据实时性要求高的场景，就比较适用与push模式。

//采用的同步发送方式，在一个线程内顺序发送,异步发送方式为：producer.send(msg, new SendCallback() {...})
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {//…}

//使用"%"操作，使得订单id取余后相同的数据路由到同一个queue中，也可以自定义路由规则
long index = id % mqs.size();  
return mqs.get((int) index);

List<PullRequest> pullRequestList = new ArrayList<PullRequest>();
  for (MessageQueue mq : mqSet) {
    if (!this.processQueueTable.containsKey(mq)) {
      if (isOrder && !this.lock(mq)) {
        log.warn("doRebalance, {}, add a new mq failed, {}, because lock failed", consumerGroup, mq);
        continue;
      }
 
     //....省略
    }
  }

public boolean lock(final MessageQueue mq) {
   FindBrokerResult findBrokerResult = this.mQClientFactory.findBrokerAddressInSubscribe(mq.getBrokerName(), MixAll.MASTER_ID, true);
   if (findBrokerResult != null) {
     LockBatchRequestBody requestBody = new LockBatchRequestBody();
     requestBody.setConsumerGroup(this.consumerGroup);
     requestBody.setClientId(this.mQClientFactory.getClientId());
     requestBody.getMqSet().add(mq);
 
     try {
       Set<MessageQueue> lockedMq =
         this.mQClientFactory.getMQClientAPIImpl().lockBatchMQ(findBrokerResult.getBrokerAddr(), requestBody, 1000);
       for (MessageQueue mmqq : lockedMq) {
         ProcessQueue processQueue = this.processQueueTable.get(mmqq);
         if (processQueue != null) {
           processQueue.setLocked(true);
           processQueue.setLastLockTimestamp(System.currentTimeMillis());
         }
       }
 
       boolean lockOK = lockedMq.contains(mq);
       log.info("the message queue lock {}, {} {}",
         lockOK ? "OK" : "Failed",
         this.consumerGroup,
         mq);
       return lockOK;
     } catch (Exception e) {
       log.error("lockBatchMQ exception, " + mq, e);
     }
   }
 
   return false;
 }

  

• RocketMQ官网示例  
  
• RocketMQ系列之pull(拉)消息模式（七）  
  
• RocketMQ的顺序消息（顺序消费）