[聚合文章] 分布式消息队列 RocketMQ 源码分析 —— Filtersrv

 1: // :arrow_down::arrow_down::arrow_down:【FiltersrvController.java】
 2: public boolean initialize(){
 3: // ....(省略代码)
 4: 
 5: // 固定间隔注册到Broker
 6: this.scheduledExecutorService.scheduleAtFixedRate(new Runnable() {
 7: 
 8: @Override
 9: public void run(){
10: FiltersrvController.this.registerFilterServerToBroker();
11: }
12: }, 15, 10, TimeUnit.SECONDS); // TODO edit by 芋艿：initialDelay时间太短，可能导致初始化失败。从3=》15
13: 
14: // ....(省略代码)
15: }
16: 
17: /**
18: * 注册Filtersrv 到 Broker
19: * ！！！如果注册失败，关闭Filtersrv
20: */
21: public void registerFilterServerToBroker(){
22: try {
23: RegisterFilterServerResponseHeader responseHeader =
24: this.filterServerOuterAPI.registerFilterServerToBroker(
25: this.filtersrvConfig.getConnectWhichBroker(), this.localAddr());
26: this.defaultMQPullConsumer.getDefaultMQPullConsumerImpl().getPullAPIWrapper()
27: .setDefaultBrokerId(responseHeader.getBrokerId());
28: 
29: if (null == this.brokerName) {
30: this.brokerName = responseHeader.getBrokerName();
31: }
32: 
33: log.info("register filter server<{}> to broker<{}> OK, Return: {} {}",
34: this.localAddr(),
35: this.filtersrvConfig.getConnectWhichBroker(),
36: responseHeader.getBrokerName(),
37: responseHeader.getBrokerId());
38: } catch (Exception e) {
39: log.warn("register filter server Exception", e);
40: 
41: log.warn("access broker failed, kill oneself");
42: System.exit(-1); // 异常退出
43: }
44: }

1: // :arrow_down::arrow_down::arrow_down:【DefaultMQPushConsumer.java】
2: @Override
3: public void subscribe(String topic, String fullClassName, String filterClassSource) throws MQClientException{
4: this.defaultMQPushConsumerImpl.subscribe(topic, fullClassName, filterClassSource);
5: }

 1: // :arrow_down::arrow_down::arrow_down:【MQClientInstance.java】
 2: /**
3: * 发送心跳到Broker，上传过滤类源码到Filtersrv
4: */
 5: public void sendHeartbeatToAllBrokerWithLock(){
 6: if (this.lockHeartbeat.tryLock()) {
 7: try {
 8: this.sendHeartbeatToAllBroker();
 9: this.uploadFilterClassSource();
10: } catch (final Exception e) {
11: log.error("sendHeartbeatToAllBroker exception", e);
12: } finally {
13: this.lockHeartbeat.unlock();
14: }
15: } else {
16: log.warn("lock heartBeat, but failed.");
17: }
18: }
19: 
20: /**
21: * 上传过滤类到Filtersrv
22: */
23: private void uploadFilterClassSource(){
24: Iterator<Entry<String, MQConsumerInner>> it = this.consumerTable.entrySet().iterator();
25: while (it.hasNext()) {
26: Entry<String, MQConsumerInner> next = it.next();
27: MQConsumerInner consumer = next.getValue();
28: if (ConsumeType.CONSUME_PASSIVELY == consumer.consumeType()) {
29: Set<SubscriptionData> subscriptions = consumer.subscriptions();
30: for (SubscriptionData sub : subscriptions) {
31: if (sub.isClassFilterMode() && sub.getFilterClassSource() != null) {
32: final String consumerGroup = consumer.groupName();
33: final String className = sub.getSubString();
34: final String topic = sub.getTopic();
35: final String filterClassSource = sub.getFilterClassSource();
36: try {
37: this.uploadFilterClassToAllFilterServer(consumerGroup, className, topic, filterClassSource);
38: } catch (Exception e) {
39: log.error("uploadFilterClassToAllFilterServer Exception", e);
40: }
41: }
42: }
43: }
44: }
45: }

 1: // :arrow_down::arrow_down::arrow_down:【FilterClassManager.java】
 2: /**
3: * 注册过滤类
4: *
5: * @param consumerGroup 消费分组
6: * @param topic Topic
7: * @param className 过滤类名
8: * @param classCRC 过滤类源码CRC
9: * @param filterSourceBinary 过滤类源码
10: * @return 是否注册成功
11: */
12: public boolean registerFilterClass(final String consumerGroup, final String topic,
13: final String className, final int classCRC, final byte[] filterSourceBinary){
14: final String key = buildKey(consumerGroup, topic);
15: // 判断是否要注册新的过滤类
16: boolean registerNew = false;
17: FilterClassInfo filterClassInfoPrev = this.filterClassTable.get(key);
18: if (null == filterClassInfoPrev) {
19: registerNew = true;
20: } else {
21: if (this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {
22: if (filterClassInfoPrev.getClassCRC() != classCRC && classCRC != 0) { // 类有变化
23: registerNew = true;
24: }
25: }
26: }
27: // 注册新的过滤类
28: if (registerNew) {
29: synchronized (this.compileLock) {
30: filterClassInfoPrev = this.filterClassTable.get(key);
31: if (null != filterClassInfoPrev && filterClassInfoPrev.getClassCRC() == classCRC) {
32: return true;
33: }
34: try {
35: FilterClassInfo filterClassInfoNew = new FilterClassInfo();
36: filterClassInfoNew.setClassName(className);
37: filterClassInfoNew.setClassCRC(0);
38: filterClassInfoNew.setMessageFilter(null);
39: 
40: if (this.filtersrvController.getFiltersrvConfig().isClientUploadFilterClassEnable()) {
41: String javaSource = new String(filterSourceBinary, MixAll.DEFAULT_CHARSET);
42: // 编译新的过滤类
43: Class<?> newClass = DynaCode.compileAndLoadClass(className, javaSource);
44: // 创建新的过滤类对象
45: Object newInstance = newClass.newInstance();
46: filterClassInfoNew.setMessageFilter((MessageFilter) newInstance);
47: filterClassInfoNew.setClassCRC(classCRC);
48: }
49: 
50: this.filterClassTable.put(key, filterClassInfoNew);
51: } catch (Throwable e) {
52: String info = String.format("FilterServer, registerFilterClass Exception, consumerGroup: %s topic: %s className: %s",
53: consumerGroup, topic, className);
54: log.error(info, e);
55: return false;
56: }
57: }
58: }
59: 
60: return true;
61: }

 1: // :arrow_down::arrow_down::arrow_down:【PullAPIWrapper.java】
 2: /**
3: * 拉取消息核心方法
4: *
5: * @param mq 消息嘟列
6: * @param subExpression 订阅表达式
7: * @param subVersion 订阅版本号
8: * @param offset 拉取队列开始位置
9: * @param maxNums 批量拉 取消息数量
10: * @param sysFlag 拉取系统标识
11: * @param commitOffset 提交消费进度
12: * @param brokerSuspendMaxTimeMillis broker挂起请求最大时间
13: * @param timeoutMillis 请求broker超时时间
14: * @param communicationMode 通讯模式
15: * @param pullCallback 拉取回调
16: * @return 拉取消息结果。只有通讯模式为同步时，才返回结果，否则返回null。
17: * @throws MQClientException 当寻找不到 broker 时，或发生其他client异常
18: * @throws RemotingException 当远程调用发生异常时
19: * @throws MQBrokerException 当 broker 发生异常时。只有通讯模式为同步时才会发生该异常。
20: * @throws InterruptedException 当发生中断异常时
21: */
22: protected PullResult pullKernelImpl(
23: final MessageQueue mq,
24: final String subExpression,
25: final long subVersion,
26: final long offset,
27: final int maxNums,
28: final int sysFlag,
29: final long commitOffset,
30: final long brokerSuspendMaxTimeMillis,
31: final long timeoutMillis,
32: final CommunicationMode communicationMode,
33: final PullCallback pullCallback
34: ) throws MQClientException, RemotingException, MQBrokerException, InterruptedException{
35: // // ....(省略代码)
36: // 请求拉取消息
37: if (findBrokerResult != null) {
38: // ....(省略代码)
39: // 若订阅topic使用过滤类，使用filtersrv获取消息
40: String brokerAddr = findBrokerResult.getBrokerAddr();
41: if (PullSysFlag.hasClassFilterFlag(sysFlagInner)) {
42: brokerAddr = computPullFromWhichFilterServer(mq.getTopic(), brokerAddr);
43: }
44: 
45: PullResult pullResult = this.mQClientFactory.getMQClientAPIImpl().pullMessage(
46: brokerAddr,
47: requestHeader,
48: timeoutMillis,
49: communicationMode,
50: pullCallback);
51: 
52: return pullResult;
53: }
54: 
55: // Broker信息不存在，则抛出异常
56: throw new MQClientException("The broker[" + mq.getBrokerName() + "] not exist", null);
57: }
58: 
59: /**
60: * 计算filtersrv地址。如果有多个filtersrv，随机选择一个。
61: *
62: * @param topic Topic
63: * @param brokerAddr broker地址
64: * @return filtersrv地址
65: * @throws MQClientException 当filtersrv不存在时
66: */
67: private String computPullFromWhichFilterServer(final String topic, final String brokerAddr)
68: throws MQClientException{
69: ConcurrentHashMap<String, TopicRouteData> topicRouteTable = this.mQClientFactory.getTopicRouteTable();
70: if (topicRouteTable != null) {
71: TopicRouteData topicRouteData = topicRouteTable.get(topic);
72: List<String> list = topicRouteData.getFilterServerTable().get(brokerAddr);
73: if (list != null && !list.isEmpty()) {
74: return list.get(randomNum() % list.size());
75: }
76: }
77: throw new MQClientException("Find Filter Server Failed, Broker Addr: " + brokerAddr + " topic: "
78: + topic, null);
79: }