[TOC]
消息过滤
消息过滤是指消费者不希望消费topic里的所有消息，而是只消费部分特定的消息。从topic中挑选出这些特定的消息，就是所谓的消息过滤。通过消息过滤可以实现消息的分流处理，例如生产者生产的消息，header可能都是不尽相同的，这样我们就可以编写两个或多个消费者，对不同header的消息进行针对性的处理。
Spring Cloud Stream实现消息过滤的方式主要有三种，一是使用@StreamListener注解的condition属性指定条件表达式，二是在消息的header中设置TAGS，三是使用SQL92语法进行过滤。
1、condition
这种方式适用于所有的MQ，也是最简单易用的方式。首先我们为生产者的消息设置一个header，具体的header名称和值可根据实际业务需求进行设置。这里仅作为示例，具体代码如下：
package com.zj.node.contentcenter.controller.content;
import lombok.RequiredArgsConstructor;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 生产者
*
* @author 01
* @date 2019-08-03
**/
@RestController
@RequiredArgsConstructor
public class TestProducerController {
private final Source source;
@GetMapping("/stream-send-msg")
public String streamSendMsg(String flagHeader) {
source.output().send(
MessageBuilder.withPayload("消息体")
// 设置用于过滤消息的header
.setHeader("flag-header", flagHeader)
.build()
);
return "send message success!";
}
}
然后在消费者的@StreamListener注解中指定condition属性，根据具体设置的header编写条件表达式。如下示例：
package com.zj.node.usercenter.rocketmq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.stereotype.Service;
/**
* 消费者
*
* @author 01
* @date 2019-08-10
**/
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(
value = Sink.INPUT,
// 消息过滤的条件表达式，只有满足该表达式才会进行执行该方法
condition = "headers['flag-header']=='UP'"
)
public void receive(String messageBody) {
log.info("通过stream收到了消息，messageBody = {}", messageBody);
}
}
现在我们来做个简单的测试，通过生产者投递两个消息，一个消息的flagHeader=UP，另一个消息的flagHeader=DOWN。消费者输出的日志信息如下：
image.png
2、TAGS
这种方式仅RocketMQ支持，Kafka、RabbitMQ不支持。该方式也需要为生产者生产的消息设置header，只不过这个header稍微有些特殊而已。我们基于上一小节的代码进行修改，修改后的代码如下：
package com.zj.node.contentcenter.controller.content;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 生产者
*
* @author 01
* @date 2019-08-03
**/
@RestController
@RequiredArgsConstructor
public class TestProducerController {
private final Source source;
@GetMapping("/stream-send-msg")
public String streamSendMsg() {
source.output().send(
MessageBuilder.withPayload("消息体")
// tag只能设置一个，不能设置多个
.setHeader(RocketMQHeaders.TAGS, "tag-1")
.build()
);
return "send message success!";
}
}
改造消费者
由于stream默认在Sink接口里只提供了一个input channel，为了更直观地看到不同tag的消息进入不同的channel，所以我们需要使用自定义接口，定义多个”input“，代码如下：
public interface MySink {
String T1_INPUT = "t1-input";
String T2_INPUT = "t2-input";
@Input(T1_INPUT)
SubscribableChannel input1();
@Input(T2_INPUT)
SubscribableChannel input2();
}
修改配置文件如下：
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.190.129:9876
bindings:
t1-input:
consumer:
# 表示t1-input仅消费带有tag-1的消息
tags: tag-1
t2-input:
consumer:
# 表示t2-input可以消费带有tag-2或tag-3的消息（||用于分隔不同的tag）
tags: tag-2||tag-3
bindings:
t1-input:
destination: stream-test-topic
group: binder-group1
t2-input:
destination: stream-test-topic
group: binder-group2
修改启动类的注解如下：
import com.zj.node.usercenter.rocketmq.MySink;
import org.springframework.cloud.stream.annotation.EnableBinding;
@EnableBinding({MySink.class})
...
修改消费者代码如下：
package com.zj.node.usercenter.rocketmq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;
/**
* 消费者
*
* @author 01
* @date 2019-08-10
**/
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(MySink.T1_INPUT)
public void receive1(String messageBody) {
log.info("消费了tag-1的消息，messageBody = {}", messageBody);
}
@StreamListener(MySink.T2_INPUT)
public void receive2(String messageBody) {
log.info("消费了tag-2 or tag-3 的消息，messageBody = {}", messageBody);
}
}
完成以上修改后，使用生产者投递消息，此时控制台输出如下：
2019-08-11 18:06:28.491 INFO 16172 --- [MessageThread_1] c.z.n.u.rocketmq.TestStreamConsumer : 消费了tag-1的消息，messageBody = 消息体
3、SQL92
这种方式仅RocketMQ支持，Kafka、RabbitMQ不支持，若使用该方式，则不要使用TAGS。该方式是使用SQL语法过滤消息，具体可参考官方文档：
http://rocketmq.apache.org/rocketmq/filter-messages-by-sql92-in-rocketmq/
默认情况下，RocketMQ的SQL过滤支持是关闭的，所以我们需要通过添加一些配置项进行开启。首先进入RocketMQ的安装目录，然后编辑conf/broker.conf文件：
[root@study-01 ~]# cd /usr/local/rocketmq-4.5.1/
[root@study-01 /usr/local/rocketmq-4.5.1]# vim conf/broker.conf
在文件末尾添加如下配置项：
enablePropertyFilter = true
注：添加配置后需要重启RocketMQ的Broker才会生效，命令参考：
nohup sh $ROCKETMQ_HOME/bin/mqbroker -n localhost:9876 -c $ROCKETMQ_HOME/conf/broker.conf &
改造生产者
同样基于上一小节的代码进行修改，生产者的代码如下：
package com.zj.node.contentcenter.controller.content;
import lombok.RequiredArgsConstructor;
import org.springframework.cloud.stream.messaging.Source;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
/**
* 生产者
*
* @author 01
* @date 2019-08-03
**/
@RestController
@RequiredArgsConstructor
public class TestProducerController {
private final Source source;
@GetMapping("/stream-send-msg")
public String streamSendMsg() {
source.output().send(
MessageBuilder.withPayload("消息体")
.setHeader("amount", "100")
.build()
);
return "send message success!";
}
}
改造消费者
由于有上一小节的铺垫，这里需要改动的就不大了，西安修改消费者的配置文件，如下：
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.190.129:9876
bindings:
t1-input:
consumer:
# 表示t1-input仅消费amount小于等于100的消息
sql: 'amount <= 100'
t2-input:
consumer:
# 表示t2-input仅消费amount大于100的消息
sql: 'amount > 100'
bindings:
t1-input:
destination: stream-test-topic
group: binder-group1
t2-input:
destination: stream-test-topic
group: binder-group2
最后修改消费者代码，如下：
package com.zj.node.usercenter.rocketmq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;
/**
* 消费者
*
* @author 01
* @date 2019-08-10
**/
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(MySink.T1_INPUT)
public void receive1(String messageBody) {
log.info("消费了amount <= 100的消息，messageBody = {}", messageBody);
}
@StreamListener(MySink.T2_INPUT)
public void receive2(String messageBody) {
log.info("消费了amount > 100的消息，messageBody = {}", messageBody);
}
}
完成以上修改后，使用生产者投递消息，此时控制台输出如下：
2019-08-11 18:40:30.019 INFO 9928 --- [MessageThread_1] c.z.n.u.rocketmq.TestStreamConsumer : 消费了amount <= 100的消息，messageBody = 消息体
关于tag和sql配置项的相关源码：
com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties
