package com.alibaba.rocketmq.client.impl.consumer;

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
	...

  	/**
	 * 订阅Topic
     * @param topic Topic名称
     * @param subExpression 过滤表达式
     * @throws MQClientException
     */
	public void subscribe(String topic, String subExpression) throws MQClientException {

		//将topic封装成订阅信息 SubscriptionData
		SubscriptionData subscriptionData = FilterAPI.buildSubscriptionData(this.defaultMQPushConsumer.getConsumerGroup(),
			topic, subExpression);
		//将订阅信息保存到订阅服务队列中
		this.rebalanceImpl.getSubscriptionInner().put(topic, subscriptionData);
		// Consumer刚启动的时候，mQClientFactory还未实例化，所以这里值为null。
		if(this.mQClientFactory != null) {
			//发现心跳消息以及更新Broker的订阅列表
			this.mQClientfactory.sendHeartbeatToAllBrokerWithLock();
		}
	}
}

package com.alibaba.rocketmq.client.impl.factory;

public class MQClientInstance {
	...

	// mQClientFactory.start()的时候会触发该方法
	public void sendHeartbeatToAllBrokerWithLock() {
		...
		//发送心跳检查
		this.sendHeartbeatToAllBroker();
		//更新Broker的订阅服务列表
		this.uploadFilterClassSource();
		...
	}

	public void uploadFilterClassSource() {
		....
		// 获取消费者的订阅信息
		Set<SubscriptionData> subscriptions = consumer.subscriptions();
		// 遍历一个个上传
		for (SubscriptionData sub : subscriptions) {
			this.uploadFilterClassSourceToAllFilterServer(
			consumerGroup, className, topic, filterClassSource);
		}
		...
	}

	public void uploadFilterClassSourceToAllFilterServer(
		final String consumerGroup, final String fullClassName, final String topic,
		final String filterClassSource) throws UnsupportedEncodingException {
		...

		this.mQClientAPIImpl.registerMessageFilterClass(fsAddr, consumerGroup, topic, fullClassName, ...);

		...
	}
}

package com.alibaba.rocketmq.client.impl;

public class MQClientAPIImpl {
	...

	public void registerMessageFilterClass(final String addr,
		final String consumerGroup,
		final String topic,
		final String className,
		...) {

		...

		//设置请求消息类型为REGISTER_MASTER_FILTER_CLASS
		RemotingCommand.createRequestCommand(RequestCode.REGISTER_MASTER_FILTER_CLASS, requestHeader);
		//发送消息事件到Broker，更新Broker的订阅服务列表
		this.remotingClient.invokeSync(addr, request, timeoutMillis);

		...
	}
}

package com.alibaba.rocketmq.client.impl.factory;

public class MQClientInstance {
	...

	//开启平衡服务
	this.rebalanceService.start();

	...
}

package com.alibaba.rocketmq.client.impl.consumer;

public class RebalanceService extends ServiceThread {
	...

	@Override
	public void run() {

		//定期执行doRebalance方法
		while(!this.isStoped) {
			this.waitForRunning(WaitInterval);
			this.mqClientFactory.doRebalance();
		}
	}

	...
}

package com.alibaba.rocketmq.client.impl.consumer;

public abstract class RebalanceImpl {
	...

	public  boolean updateProcessQueueTableInRebalance(final String topic, fianl Set<MessageQueue> mqSet) {
		...

		//1.移除过期无用的MessageQueue
		if(!mqSet.contains(mq)) {
			this.removeUnnecessaryMessageQueue(mq, pq);
		} else if (pq.isPullExpired) {
			this.removeUnnecessaryMessageQueue(mq, pq);
		}
		//2.将新增的MessageQueue加入ProcessQueueTable消息处理队列集合中
		this.processQueueTable.put(mq, pullRequest.getProcessQueue());
		//3.将MessageQueue封装成PullRequest请求，然后发起请求
		this.dispatchPullRequest(pullRequestList);
	}

	...
}

package com.alibaba.rocketmq.client.impl.consumer;

public class PullMessageService extends ServiceThread {
	...

	public void executePullRequestImmediately(fianl PullRequest pullRequest) {
		//将请求加入请求队列中
		this.pullRequestQueue.put(pullRequest);
	}

	@Override
	public void run() {

		while(!this.isStoped) {
			//一直轮询监听请求队列
			PullRequest pullRequest = this.pullRequestQueue.take();
			if(pullRequest != null) {
				//当一有新请求加入队列时，就处理该请求
				this.pullMessage(pullReuqest);
			}
		}
	}
}

package com.alibaba.rocketmq.client.impl.consumer;

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
	...

	/**
     * 消费者注册自定义消息处理程序，当接收到新消息时会回调用该处理程序
     * 处理消息。
     * @param messageListener
     */
    public void registerMessageListener(MessageListener messageListener) {
        this.messageListenerInner = messageListener;
    }

    public void pullMessage(final PullRequest pullRequest) {

    	...

    	//消息请求回调方法
    	PullCallback pullCallback = new PullCallback() {

    		@Override
    		public void onSuccess(PullResult pullRequest) {
    			...

    			DefaultMQPushConsumerImpl.this.consumeMessageService.submitConsumeRequest(
    				pullResult.getMsgFoundList(),
    				processQueue,
    				pullRequest.getMessageQueue(),
    				dispathToConsume
    				);

    			...
    		}

    	}

    	...
    }
}

package com.alibaba.rocketmq.client.impl.consumer;

public class ConsumeMessageConcurrentlyService implements ConsumeMessageService {
	...

	@Override
	public void submitConsumeRequest(final List<MessageExt> msgs, ...) {
		...
		//1.将消息响应封装成一个Task任务
		ConsumeRequest consumeRequest = new ConsumeRequest(msgs, processQueue, messageQueue);
		//2.将该任务提交到线程池
		this.consumeExecutor.submit(consumeRequest);
		...
	}

	class ConsumeRequest implements Runnable {
		...

		@Override
		public void run() {
			//用户自定义的消息处理回调方法
			MessageListenerConcurrently listener = ConsumeMessageConcurrentlyService.this.messageListener;
			//调用用户自定义的方法
			status = listener.consumeMessage(Collections.unmodifiableList(msgs), context);
		}

		...

	}
}

package com.alibaba.rocketmq.client.impl.consumer;

public class DefaultMQPushConsumerImpl implements MQConsumerInner {
	...

	// 消费进度存储
	private OffsetStore offsetStore;

	public void start() throws MQClientException {
		...

		if(this.defaultMQPushConsumer.getOffsetStore() != null) {
			this.offsetStore = this.defaultMQPushConsumer.getOffsetStore();
		} else {
			// 广播消费/集群消费
            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;
            }
		}

		// 加载消费进度
        this.offsetStore.load();

        ...
	}
}

{
	"offsetTable":{
		"%RETRY%PushConsumer@PushConsumer":{0:0
		},
		"%RETRY%PushConsumer-1@PushConsumer-1":{0:0
		},
		// PushTopic-5 主题 PushConsumer 消费者组别 0,1,2,3 表示消息队列（默认有四个队列）
		"PushTopic-5@PushConsumer":{0:2,1:3,2:0,3:0 // {0:2,...} 0 表示第一个消息队列，2表示该消费者消费的进度
		},
		"PushTopic-5@PushConsumer-1":{0:2,1:1,2:0,3:0
		},
		"PushTopic@PushConsumer":{0:0,1:0,2:1,3:1
		}
	}
}

package com.alibaba.rocketmq.broker.processor;

public class ClientManageProcessor implements NettyRequestProcessor {
	...

	@Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request)
            throws RemotingCommandException {
        switch (request.getCode()) {
        case RequestCode.HEART_BEAT: // 心跳请求
            return this.heartBeat(ctx, request);
        case RequestCode.UNREGISTER_CLIENT: // 注销客户端
            return this.unregisterClient(ctx, request);
        case RequestCode.GET_CONSUMER_LIST_BY_GROUP: // 获取消费者列表
            return this.getConsumerListByGroup(ctx, request);
        case RequestCode.UPDATE_CONSUMER_OFFSET: // 更新消费进度
            return this.updateConsumerOffset(ctx, request);
        case RequestCode.QUERY_CONSUMER_OFFSET: // 查询消费进度
            return this.queryConsumerOffset(ctx, request);
        default:
            break;  
        }
        return null;
    }

	/**
	 * 更新消费进度
	 * @param ctx
	 * @param request
	 * @return
	 * @throws RemotingCommandException
	 */
 	private RemotingCommand updateConsumerOffset(ChannelHandlerContext ctx, RemotingCommand request)
            throws RemotingCommandException {
        ...

        // 反序列化请求
        final UpdateConsumerOffsetRequestHeader requestHeader =
                (UpdateConsumerOffsetRequestHeader) request
                    .decodeCommandCustomHeader(UpdateConsumerOffsetRequestHeader.class);

        // 更新消费进度
        this.brokerController.getConsumerOffsetManager().commitOffset(requestHeader.getConsumerGroup(),
            requestHeader.getTopic(), requestHeader.getQueueId(), requestHeader.getCommitOffset());

        ...
    }
}

package com.alibaba.rocketmq.broker.offset;

/**
 * Consumer消费进度管理
 * 
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-8-11
 */
public class ConsumerOffsetManager extends ConfigManager {

	private static final String TOPIC_GROUP_SEPARATOR = "@";

	// 消费进度队列
	private ConcurrentHashMap<String/* topic@group */, ConcurrentHashMap<Integer, Long>> offsetTable =
            new ConcurrentHashMap<String, ConcurrentHashMap<Integer, Long>>(512);

	/**
	 * 更新消费进度
	 * @param group 消费者组别
	 * @param topic 消息主题
	 * @param queueId 消息队列ID
	 * @param offset 消费进度
	 */
    public void commitOffset(final String group, final String topic, final int queueId, final long offset) {
        // topic@group 例如 配置文件中保存的 "PushTopic-5@PushConsumer"
        String key = topic + TOPIC_GROUP_SEPARATOR + group;
        this.commitOffset(key, queueId, offset);
    }

    private void commitOffset(final String key, final int queueId, final long offset) {
        ConcurrentHashMap<Integer, Long> map = this.offsetTable.get(key);
        if (null == map) {
            map = new ConcurrentHashMap<Integer, Long>(32);
            map.put(queueId, offset);
            this.offsetTable.put(key, map); // 保存到队列中
        }
        else {
            map.put(queueId, offset);
        }
    }

	/**
	 * 查询消费进度
	 * @param group 消费者组别
	 * @param topic 消息主题
	 * @param queueId 消息队列ID
	 */
    public long queryOffset(final String group, final String topic, final int queueId) {
        // topic@group 根据 "PushTopic-5@PushConsumer" 来查询
        String key = topic + TOPIC_GROUP_SEPARATOR + group;
        ConcurrentHashMap<Integer, Long> map = this.offsetTable.get(key);
        if (null != map) {
            Long offset = map.get(queueId);
            if (offset != null)
                return offset;
        }

        return -1;
    }
}