@Test public void testDispatch() { BrokerConfig brokerConfig = new BrokerConfig(); brokerConfig.setEnableCalcFilterBitMap(true); ConsumerFilterManager filterManager = ConsumerFilterManagerTest.gen(10, 10); CommitLogDispatcherCalcBitMap calcBitMap = new CommitLogDispatcherCalcBitMap(brokerConfig, filterManager); for (int i = 0; i < 10; i++) { Map<String, String> properties = new HashMap<String, String>(4); properties.put("a", String.valueOf(i * 10 + 5)); String topic = "topic" + i; DispatchRequest dispatchRequest = new DispatchRequest( topic, 0, i * 100 + 123, 100, (long) ("tags" + i).hashCode(), System.currentTimeMillis(), i, null, UUID.randomUUID().toString(), 0, 0, properties ); calcBitMap.dispatch(dispatchRequest); assertThat(dispatchRequest.getBitMap()).isNotNull(); BitsArray bits = BitsArray.create(dispatchRequest.getBitMap()); Collection<ConsumerFilterData> filterDatas = filterManager.get(topic); for (ConsumerFilterData filterData : filterDatas) { if (filterManager.getBloomFilter().isHit(filterData.getBloomFilterData(), bits)) { try { assertThat((Boolean) filterData.getCompiledExpression().evaluate( new MessageEvaluationContext(properties) )).isTrue(); } catch (Exception e) { e.printStackTrace(); assertThat(true).isFalse(); } } else { try { assertThat((Boolean) filterData.getCompiledExpression().evaluate( new MessageEvaluationContext(properties) )).isFalse(); } catch (Exception e) { e.printStackTrace(); assertThat(true).isFalse(); } } } } }
@Override public void dispatch(DispatchRequest request) { if (!this.brokerConfig.isEnableCalcFilterBitMap()) { return; } try { Collection<ConsumerFilterData> filterDatas = consumerFilterManager.get(request.getTopic()); if (filterDatas == null || filterDatas.isEmpty()) { return; } Iterator<ConsumerFilterData> iterator = filterDatas.iterator(); BitsArray filterBitMap = BitsArray.create( this.consumerFilterManager.getBloomFilter().getM() ); long startTime = System.currentTimeMillis(); while (iterator.hasNext()) { ConsumerFilterData filterData = iterator.next(); if (filterData.getCompiledExpression() == null) { log.error("[BUG] Consumer in filter manager has no compiled expression! {}", filterData); continue; } if (filterData.getBloomFilterData() == null) { log.error("[BUG] Consumer in filter manager has no bloom data! {}", filterData); continue; } Object ret = null; try { MessageEvaluationContext context = new MessageEvaluationContext(request.getPropertiesMap()); ret = filterData.getCompiledExpression().evaluate(context); } catch (Throwable e) { log.error("Calc filter bit map error!commitLogOffset={}, consumer={}, {}", request.getCommitLogOffset(), filterData, e); } log.debug("Result of Calc bit map:ret={}, data={}, props={}, offset={}", ret, filterData, request.getPropertiesMap(), request.getCommitLogOffset()); if (ret != null && ret instanceof Boolean && (Boolean) ret) { consumerFilterManager.getBloomFilter().hashTo( filterData.getBloomFilterData(), filterBitMap ); } } request.setBitMap(filterBitMap.bytes()); long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(startTime); if (elapsedTime >= 1) { log.warn("Spend {} ms to calc bit map, consumerNum={}, topic={}", elapsedTime, filterDatas.size(), request.getTopic()); } } catch (Throwable e) { log.error("Calc bit map error! topic={}, offset={}, queueId={}, {}", request.getTopic(), request.getCommitLogOffset(), request.getQueueId(), e); } }
CommitLogDispatcherCalcBitMap implements CommitLogDispatcher { @Override public void dispatch(DispatchRequest request) { if (!this.brokerConfig.isEnableCalcFilterBitMap()) { return; } try { Collection<ConsumerFilterData> filterDatas = consumerFilterManager.get(request.getTopic()); if (filterDatas == null || filterDatas.isEmpty()) { return; } Iterator<ConsumerFilterData> iterator = filterDatas.iterator(); BitsArray filterBitMap = BitsArray.create( this.consumerFilterManager.getBloomFilter().getM() ); long startTime = System.currentTimeMillis(); while (iterator.hasNext()) { ConsumerFilterData filterData = iterator.next(); if (filterData.getCompiledExpression() == null) { log.error("[BUG] Consumer in filter manager has no compiled expression! {}", filterData); continue; } if (filterData.getBloomFilterData() == null) { log.error("[BUG] Consumer in filter manager has no bloom data! {}", filterData); continue; } Object ret = null; try { MessageEvaluationContext context = new MessageEvaluationContext(request.getPropertiesMap()); ret = filterData.getCompiledExpression().evaluate(context); } catch (Throwable e) { log.error("Calc filter bit map error!commitLogOffset={}, consumer={}, {}", request.getCommitLogOffset(), filterData, e); } log.debug("Result of Calc bit map:ret={}, data={}, props={}, offset={}", ret, filterData, request.getPropertiesMap(), request.getCommitLogOffset()); if (ret != null && ret instanceof Boolean && (Boolean) ret) { consumerFilterManager.getBloomFilter().hashTo( filterData.getBloomFilterData(), filterBitMap ); } } request.setBitMap(filterBitMap.bytes()); long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(startTime); if (elapsedTime >= 1) { log.warn("Spend {} ms to calc bit map, consumerNum={}, topic={}", elapsedTime, filterDatas.size(), request.getTopic()); } } catch (Throwable e) { log.error("Calc bit map error! topic={}, offset={}, queueId={}, {}", request.getTopic(), request.getCommitLogOffset(), request.getQueueId(), e); } } }
CommitLogDispatcherCalcBitMap implements CommitLogDispatcher { @Override public void dispatch(DispatchRequest request) { if (!this.brokerConfig.isEnableCalcFilterBitMap()) { return; } try { Collection<ConsumerFilterData> filterDatas = consumerFilterManager.get(request.getTopic()); if (filterDatas == null || filterDatas.isEmpty()) { return; } Iterator<ConsumerFilterData> iterator = filterDatas.iterator(); BitsArray filterBitMap = BitsArray.create( this.consumerFilterManager.getBloomFilter().getM() ); long startTime = System.currentTimeMillis(); while (iterator.hasNext()) { ConsumerFilterData filterData = iterator.next(); if (filterData.getCompiledExpression() == null) { log.error("[BUG] Consumer in filter manager has no compiled expression! {}", filterData); continue; } if (filterData.getBloomFilterData() == null) { log.error("[BUG] Consumer in filter manager has no bloom data! {}", filterData); continue; } Object ret = null; try { MessageEvaluationContext context = new MessageEvaluationContext(request.getPropertiesMap()); ret = filterData.getCompiledExpression().evaluate(context); } catch (Throwable e) { log.error("Calc filter bit map error!commitLogOffset={}, consumer={}, {}", request.getCommitLogOffset(), filterData, e); } log.debug("Result of Calc bit map:ret={}, data={}, props={}, offset={}", ret, filterData, request.getPropertiesMap(), request.getCommitLogOffset()); if (ret != null && ret instanceof Boolean && (Boolean) ret) { consumerFilterManager.getBloomFilter().hashTo( filterData.getBloomFilterData(), filterBitMap ); } } request.setBitMap(filterBitMap.bytes()); long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(startTime); if (elapsedTime >= 1) { log.warn("Spend {} ms to calc bit map, consumerNum={}, topic={}", elapsedTime, filterDatas.size(), request.getTopic()); } } catch (Throwable e) { log.error("Calc bit map error! topic={}, offset={}, queueId={}, {}", request.getTopic(), request.getCommitLogOffset(), request.getQueueId(), e); } } CommitLogDispatcherCalcBitMap(BrokerConfig brokerConfig, ConsumerFilterManager consumerFilterManager); }
CommitLogDispatcherCalcBitMap implements CommitLogDispatcher { @Override public void dispatch(DispatchRequest request) { if (!this.brokerConfig.isEnableCalcFilterBitMap()) { return; } try { Collection<ConsumerFilterData> filterDatas = consumerFilterManager.get(request.getTopic()); if (filterDatas == null || filterDatas.isEmpty()) { return; } Iterator<ConsumerFilterData> iterator = filterDatas.iterator(); BitsArray filterBitMap = BitsArray.create( this.consumerFilterManager.getBloomFilter().getM() ); long startTime = System.currentTimeMillis(); while (iterator.hasNext()) { ConsumerFilterData filterData = iterator.next(); if (filterData.getCompiledExpression() == null) { log.error("[BUG] Consumer in filter manager has no compiled expression! {}", filterData); continue; } if (filterData.getBloomFilterData() == null) { log.error("[BUG] Consumer in filter manager has no bloom data! {}", filterData); continue; } Object ret = null; try { MessageEvaluationContext context = new MessageEvaluationContext(request.getPropertiesMap()); ret = filterData.getCompiledExpression().evaluate(context); } catch (Throwable e) { log.error("Calc filter bit map error!commitLogOffset={}, consumer={}, {}", request.getCommitLogOffset(), filterData, e); } log.debug("Result of Calc bit map:ret={}, data={}, props={}, offset={}", ret, filterData, request.getPropertiesMap(), request.getCommitLogOffset()); if (ret != null && ret instanceof Boolean && (Boolean) ret) { consumerFilterManager.getBloomFilter().hashTo( filterData.getBloomFilterData(), filterBitMap ); } } request.setBitMap(filterBitMap.bytes()); long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(startTime); if (elapsedTime >= 1) { log.warn("Spend {} ms to calc bit map, consumerNum={}, topic={}", elapsedTime, filterDatas.size(), request.getTopic()); } } catch (Throwable e) { log.error("Calc bit map error! topic={}, offset={}, queueId={}, {}", request.getTopic(), request.getCommitLogOffset(), request.getQueueId(), e); } } CommitLogDispatcherCalcBitMap(BrokerConfig brokerConfig, ConsumerFilterManager consumerFilterManager); @Override void dispatch(DispatchRequest request); }
CommitLogDispatcherCalcBitMap implements CommitLogDispatcher { @Override public void dispatch(DispatchRequest request) { if (!this.brokerConfig.isEnableCalcFilterBitMap()) { return; } try { Collection<ConsumerFilterData> filterDatas = consumerFilterManager.get(request.getTopic()); if (filterDatas == null || filterDatas.isEmpty()) { return; } Iterator<ConsumerFilterData> iterator = filterDatas.iterator(); BitsArray filterBitMap = BitsArray.create( this.consumerFilterManager.getBloomFilter().getM() ); long startTime = System.currentTimeMillis(); while (iterator.hasNext()) { ConsumerFilterData filterData = iterator.next(); if (filterData.getCompiledExpression() == null) { log.error("[BUG] Consumer in filter manager has no compiled expression! {}", filterData); continue; } if (filterData.getBloomFilterData() == null) { log.error("[BUG] Consumer in filter manager has no bloom data! {}", filterData); continue; } Object ret = null; try { MessageEvaluationContext context = new MessageEvaluationContext(request.getPropertiesMap()); ret = filterData.getCompiledExpression().evaluate(context); } catch (Throwable e) { log.error("Calc filter bit map error!commitLogOffset={}, consumer={}, {}", request.getCommitLogOffset(), filterData, e); } log.debug("Result of Calc bit map:ret={}, data={}, props={}, offset={}", ret, filterData, request.getPropertiesMap(), request.getCommitLogOffset()); if (ret != null && ret instanceof Boolean && (Boolean) ret) { consumerFilterManager.getBloomFilter().hashTo( filterData.getBloomFilterData(), filterBitMap ); } } request.setBitMap(filterBitMap.bytes()); long elapsedTime = UtilAll.computeElapsedTimeMilliseconds(startTime); if (elapsedTime >= 1) { log.warn("Spend {} ms to calc bit map, consumerNum={}, topic={}", elapsedTime, filterDatas.size(), request.getTopic()); } } catch (Throwable e) { log.error("Calc bit map error! topic={}, offset={}, queueId={}, {}", request.getTopic(), request.getCommitLogOffset(), request.getQueueId(), e); } } CommitLogDispatcherCalcBitMap(BrokerConfig brokerConfig, ConsumerFilterManager consumerFilterManager); @Override void dispatch(DispatchRequest request);  }