SpringCloudStream实现消息过滤的三种主要方式
消息过滤
消息过滤是指消费者不希望消费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
。消费者输出的日志信息如下:
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
分享名称:SpringCloudStream实现消息过滤的三种主要方式
文章源于:http://myzitong.com/article/piipdg.html