SpringCloudStream實(shí)現(xiàn)消息過濾的三種主要方式

消息過濾

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

在新化等地區(qū)，都構(gòu)建了全面的區(qū)域性戰(zhàn)略布局，加強(qiáng)發(fā)展的系統(tǒng)性、市場前瞻性、產(chǎn)品創(chuàng)新能力，以專注、極致的服務(wù)理念，為客戶提供做網(wǎng)站、網(wǎng)站設(shè)計(jì) 網(wǎng)站設(shè)計(jì)制作按需求定制開發(fā),公司網(wǎng)站建設(shè),企業(yè)網(wǎng)站建設(shè),高端網(wǎng)站設(shè)計(jì),全網(wǎng)整合營銷推廣,外貿(mào)網(wǎng)站制作,新化網(wǎng)站建設(shè)費(fèi)用合理。

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á)式才會進(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)者輸出的日志信息如下：
Spring Cloud Stream實(shí)現(xiàn)消息過濾的三種主要方式

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

修改啟動類的注解如下：

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í)控制臺輸出如下：

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才會生效，命令參考：

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é)的鋪墊，這里需要改動的就不大了，首先修改消費(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í)控制臺輸出如下：

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

名稱欄目：SpringCloudStream實(shí)現(xiàn)消息過濾的三種主要方式
當(dāng)前路徑：http://www.chinadenli.net/article16/piipdg.html

成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián)，為您提供企業(yè)建站、電子商務(wù)、外貿(mào)網(wǎng)站建設(shè)、網(wǎng)站策劃、建站公司、軟件開發(fā)

聲明：本網(wǎng)站發(fā)布的內(nèi)容（圖片、視頻和文字）以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主，如果涉及侵權(quán)請盡快告知，我們將會在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場，如需處理請聯(lián)系客服。電話：028-86922220；郵箱：631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載，或轉(zhuǎn)載時(shí)需注明來源：創(chuàng)新互聯(lián)

猜你還喜歡下面的內(nèi)容

欧美一区二区三区老妇人-欧美做爰猛烈大尺度电-99久久夜色精品国产亚洲a-亚洲福利视频一区二区

SpringCloudStream實(shí)現(xiàn)消息過濾的三種主要方式

消息過濾

1、condition

2、TAGS

3、SQL92