1、过滤器示例
在大多数情况下，tag是一种简单实用的设计，用于选择所需的消息。例如：
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("CID_EXAMPLE");
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");
消费者将收到包含TAGA或TAGB或TAGC的消息。但限制是一条消息只能有一个标记，这可能不适用于复杂的方案。在这种情况下，您可以使用SQL表达式过滤掉消息。
原理
SQL功能可以通过发送消息时输入的属性进行一些计算。在RocketMQ定义的语法下，您可以实现一些有趣的逻辑。这是一个例子：
image.png
文法
RocketMQ仅定义了一些支持此功能的基本语法。你也可以轻松扩展它。
- 数值比较，如
>，>=，<，<=，BETWEEN，=; - 字符比较，如
=，<>，IN; -
IS NULL或IS NOT NULL; - 逻辑
AND，OR，NOT，
常量类型是：
- 数字，如123,3.1415;
- 字符，如'abc'，必须用单引号制作;
-
NULL，特殊常数; - 布尔值，
TRUE或FALSE;
使用限制
只有push consumer可以通过SQL92选择消息。界面是：
public void subscribe(final String topic, final MessageSelector messageSelector)
使用注意：默认情况下rocketMQ不支持SQL92 需要修改配置，配置位于
rocketmq-all-4.4.0/distribution/target/apache-rocketmq/conf，添加 enablePropertyFilter=true
启动的时候加上配置：
nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf autoCreateTopicEnable=true &
Producer example
您可以在发送时通过方法将属性放入消息中putUserProperty。
public class FilterProducer {
private static int i=2;
public static void main(String[] args) throws Exception {
DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
producer.setNamesrvAddr("192.168.247.132:9876");
producer.start();
Message msg = new Message("TopicTest",
"Tag A",
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET)
);
// Set some properties.
msg.putUserProperty("a", String.valueOf(i));
SendResult sendResult = producer.send(msg);
producer.shutdown();
}
}
Consumer example
用于MessageSelector.bySql在使用时通过SQL92选择消息。
public class FilterConsumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_4");
consumer.setNamesrvAddr("192.168.247.132:9876");
// only subsribe messages have property a, also a >=0 and a <= 3
consumer.subscribe("TopicTest", MessageSelector.bySql("a <= 10"));
consumer.registerMessageListener(new MessageListenerConcurrently() {
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
for (MessageExt msg : msgs) {
System.out.println(msg.getMsgId()+" "+msg);
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
}
}
2、Logappender示例
以日志的形式走消息队列
RocketMQ logappender提供log4j appender，log4j2 appender和logback appender供商务使用，下面是配置示例。
log4j的
使用log4j属性配置文件时，配置如下。
log4j.appender.mq=org.apache.rocketmq.logappender.log4j.RocketmqLog4jAppender
log4j.appender.mq.Tag=yourTag
log4j.appender.mq.Topic=yourLogTopic
log4j.appender.mq.ProducerGroup=yourLogGroup
log4j.appender.mq.NameServerAddress=yourRocketmqNameserverAddress
log4j.appender.mq.layout=org.apache.log4j.PatternLayout
log4j.appender.mq.layout.ConversionPattern=%d{yyyy-MM-dd HH:mm:ss} %-4r [%t] (%F:%L) %-5p - %m%n
使用log4j xml配置文件时，请将其配置为此并添加异步appender：
<appender name="mqAppender1" class="org.apache.rocketmq.logappender.log4j.RocketmqLog4jAppender">
<param name="Tag" value="yourTag" />
<param name="Topic" value="yourLogTopic" />
<param name="ProducerGroup" value="yourLogGroup" />
<param name="NameServerAddress" value="yourRocketmqNameserverAddress"/>
<layout class="org.apache.log4j.PatternLayout">
<param name="ConversionPattern" value="%d{yyyy-MM-dd HH:mm:ss}-%p %t %c - %m%n" />
</layout>
</appender>
<appender name="mqAsyncAppender1" class="org.apache.log4j.AsyncAppender">
<param name="BufferSize" value="1024" />
<param name="Blocking" value="false" />
<appender-ref ref="mqAppender1"/>
</appender>
log4j2
当使用log4j2时，如此配置。如果你想要noneblock，只需为ref配置一个asyncAppender。
<RocketMQ name="rocketmqAppender" producerGroup="yourLogGroup" nameServerAddress="yourRocketmqNameserverAddress"
topic="yourLogTopic" tag="yourTag">
<PatternLayout pattern="%d [%p] hahahah %c %m%n"/>
</RocketMQ>
的logback
使用logback时，还需要asyncAppender。
<appender name="mqAppender1" class="org.apache.rocketmq.logappender.logback.RocketmqLogbackAppender">
<tag>yourTag</tag>
<topic>yourLogTopic</topic>
<producerGroup>yourLogGroup</producerGroup>
<nameServerAddress>yourRocketmqNameserverAddress</nameServerAddress>
<layout>
<pattern>%date %p %t - %m%n</pattern>
</layout>
</appender>
<appender name="mqAsyncAppender1" class="ch.qos.logback.classic.AsyncAppender">
<queueSize>1024</queueSize>
<discardingThreshold>80</discardingThreshold>
<maxFlushTime>2000</maxFlushTime>
<neverBlock>true</neverBlock>
<appender-ref ref="mqAppender1"/>
</appender>
