[TOC]
消息過濾
消息過濾是指消費(fèi)者不希望消費(fèi)topic里的所有消息漆枚，而是只消費(fèi)部分特定的消息。從topic中挑選出這些特定的消息抵知，就是所謂的消息過濾墙基。通過消息過濾可以實(shí)現(xiàn)消息的分流處理软族，例如生產(chǎn)者生產(chǎn)的消息，header可能都是不盡相同的残制，這樣我們就可以編寫兩個(gè)或多個(gè)消費(fèi)者立砸，對(duì)不同header的消息進(jìn)行針對(duì)性的處理。
Spring Cloud Stream實(shí)現(xiàn)消息過濾的方式主要有三種初茶，一是使用@StreamListener注解的condition屬性指定條件表達(dá)式颗祝，二是在消息的header中設(shè)置TAGS，三是使用SQL92語法進(jìn)行過濾恼布。
1螺戳、condition
這種方式適用于所有的MQ，也是最簡單易用的方式折汞。首先我們?yōu)樯a(chǎn)者的消息設(shè)置一個(gè)header倔幼，具體的header名稱和值可根據(jù)實(shí)際業(yè)務(wù)需求進(jìn)行設(shè)置。這里僅作為示例爽待，具體代碼如下：
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;
/**
* 生產(chǎn)者
*
* @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("消息體")
// 設(shè)置用于過濾消息的header
.setHeader("flag-header", flagHeader)
.build()
);
return "send message success!";
}
}
然后在消費(fèi)者的@StreamListener注解中指定condition屬性损同，根據(jù)具體設(shè)置的header編寫條件表達(dá)式。如下示例：
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;
/**
* 消費(fèi)者
*
* @author 01
* @date 2019-08-10
**/
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(
value = Sink.INPUT,
// 消息過濾的條件表達(dá)式鸟款，只有滿足該表達(dá)式才會(huì)進(jìn)行執(zhí)行該方法
condition = "headers['flag-header']=='UP'"
)
public void receive(String messageBody) {
log.info("通過stream收到了消息膏燃，messageBody = {}", messageBody);
}
}
現(xiàn)在我們來做個(gè)簡單的測試，通過生產(chǎn)者投遞兩個(gè)消息欠雌，一個(gè)消息的flagHeader=UP蹄梢，另一個(gè)消息的flagHeader=DOWN。消費(fèi)者輸出的日志信息如下：
2富俄、TAGS
這種方式僅RocketMQ支持禁炒，Kafka、RabbitMQ不支持霍比。該方式也需要為生產(chǎn)者生產(chǎn)的消息設(shè)置header幕袱，只不過這個(gè)header稍微有些特殊而已。我們基于上一小節(jié)的代碼進(jìn)行修改悠瞬，修改后的代碼如下：
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;
/**
* 生產(chǎn)者
*
* @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只能設(shè)置一個(gè)们豌，不能設(shè)置多個(gè)
.setHeader(RocketMQHeaders.TAGS, "tag-1")
.build()
);
return "send message success!";
}
}
改造消費(fèi)者
由于stream默認(rèn)在Sink接口里只提供了一個(gè)input channel，為了更直觀地看到不同tag的消息進(jìn)入不同的channel浅妆，所以我們需要使用自定義接口望迎，定義多個(gè)”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僅消費(fèi)帶有tag-1的消息
tags: tag-1
t2-input:
consumer:
# 表示t2-input可以消費(fèi)帶有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
修改啟動(dòng)類的注解如下：
import com.zj.node.usercenter.rocketmq.MySink;
import org.springframework.cloud.stream.annotation.EnableBinding;
@EnableBinding({MySink.class})
...
修改消費(fèi)者代碼如下：
package com.zj.node.usercenter.rocketmq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;
/**
* 消費(fèi)者
*
* @author 01
* @date 2019-08-10
**/
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(MySink.T1_INPUT)
public void receive1(String messageBody) {
log.info("消費(fèi)了tag-1的消息凌外，messageBody = {}", messageBody);
}
@StreamListener(MySink.T2_INPUT)
public void receive2(String messageBody) {
log.info("消費(fèi)了tag-2 or tag-3 的消息辩尊，messageBody = {}", messageBody);
}
}
完成以上修改后，使用生產(chǎn)者投遞消息康辑，此時(shí)控制臺(tái)輸出如下：
2019-08-11 18:06:28.491 INFO 16172 --- [MessageThread_1] c.z.n.u.rocketmq.TestStreamConsumer : 消費(fèi)了tag-1的消息摄欲，messageBody = 消息體
3轿亮、SQL92
這種方式僅RocketMQ支持，Kafka胸墙、RabbitMQ不支持我注，若使用該方式，則不要使用TAGS迟隅。該方式是使用SQL語法過濾消息但骨，具體可參考官方文檔：
http://rocketmq.apache.org/rocketmq/filter-messages-by-sql92-in-rocketmq/
默認(rèn)情況下，RocketMQ的SQL過濾支持是關(guān)閉的智袭，所以我們需要通過添加一些配置項(xiàng)進(jìn)行開啟嗽冒。首先進(jìn)入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
在文件末尾添加如下配置項(xiàng)：
enablePropertyFilter = true
注：添加配置后需要重啟RocketMQ的Broker才會(huì)生效补履，命令參考：
nohup sh $ROCKETMQ_HOME/bin/mqbroker -n localhost:9876 -c $ROCKETMQ_HOME/conf/broker.conf &
改造生產(chǎn)者
同樣基于上一小節(jié)的代碼進(jìn)行修改添坊，生產(chǎn)者的代碼如下：
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;
/**
* 生產(chǎn)者
*
* @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!";
}
}
改造消費(fèi)者
由于有上一小節(jié)的鋪墊，這里需要改動(dòng)的就不大了箫锤，西安修改消費(fèi)者的配置文件贬蛙，如下：
spring:
cloud:
stream:
rocketmq:
binder:
name-server: 192.168.190.129:9876
bindings:
t1-input:
consumer:
# 表示t1-input僅消費(fèi)amount小于等于100的消息
sql: 'amount <= 100'
t2-input:
consumer:
# 表示t2-input僅消費(fèi)amount大于100的消息
sql: 'amount > 100'
bindings:
t1-input:
destination: stream-test-topic
group: binder-group1
t2-input:
destination: stream-test-topic
group: binder-group2
最后修改消費(fèi)者代碼，如下：
package com.zj.node.usercenter.rocketmq;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.stereotype.Service;
/**
* 消費(fèi)者
*
* @author 01
* @date 2019-08-10
**/
@Slf4j
@Service
public class TestStreamConsumer {
@StreamListener(MySink.T1_INPUT)
public void receive1(String messageBody) {
log.info("消費(fèi)了amount <= 100的消息谚攒，messageBody = {}", messageBody);
}
@StreamListener(MySink.T2_INPUT)
public void receive2(String messageBody) {
log.info("消費(fèi)了amount > 100的消息阳准，messageBody = {}", messageBody);
}
}
完成以上修改后，使用生產(chǎn)者投遞消息馏臭，此時(shí)控制臺(tái)輸出如下：
2019-08-11 18:40:30.019 INFO 9928 --- [MessageThread_1] c.z.n.u.rocketmq.TestStreamConsumer : 消費(fèi)了amount <= 100的消息野蝇，messageBody = 消息體
關(guān)于tag和sql配置項(xiàng)的相關(guān)源碼：
com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties
