Spring Cloud Stream實現消息過濾的三種主要方式

消息過濾

消息過濾是指消費者不希望消費topic里的所有消息，而是只消費部分特定的消息。從topic中挑選出這些特定的消息，就是所謂的消息過濾。通過消息過濾可以實現消息的分流處理，例如生產者生產的消息，header可能都是不盡相同的，這樣我們就可以編寫兩個或多個消費者，對不同header的消息進行針對性的處理。

Spring Cloud Stream實現消息過濾的方式主要有三種，一是使用@StreamListener注解的condition屬性指定條件表達式，二是在消息的header中設置TAGS，三是使用SQL92語法進行過濾。

1、condition

這種方式適用于所有的MQ，也是最簡單易用的方式。首先我們?yōu)樯a者的消息設置一個header，具體的header名稱和值可根據實際業(yè)務需求進行設置。這里僅作為示例，具體代碼如下：

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,
            // 消息過濾的條件表達式，只有滿足該表達式才會進行執(zhí)行該方法
            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稍微有些特殊而已。我們基于上一小節(jié)的代碼進行修改，修改后的代碼如下：

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 &

改造生產者

同樣基于上一小節(jié)的代碼進行修改，生產者的代碼如下：

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!";
    }
}

改造消費者

由于有上一小節(jié)的鋪墊，這里需要改動的就不大了，首先修改消費者的配置文件，如下：

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