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>

五、RocketMQ案例（二）