下载pulsar-msg-filter-plugin-VERSION.nar插件并保存至指定目录，如/app/conf/plugin

修改pulsar broker.conf配置（version >= 2.10），插件名称pulsar-msg-filter

# Class name of Pluggable entry filter that can decide whether the entry needs to be filtered
# You can use this class to decide which entries can be sent to consumers.
# Multiple classes need to be separated by commas.
entryFilterNames=pulsar-msg-filter

# The directory for all the entry filter implementations
entryFiltersDirectory=/app/plugin
# Location of unpacked NAR file
narExtractionDirectory=/app/nar

重启broker，查看日志，如果看到如下日志：

Successfully loaded entry filter for name `pulsar-msg-filter`

则说明配置成功

验证（option）

1. 发送方构建Producer实例时关闭 batch 操作 .enableBatching(false)

   Producer<String> producer = client.newProducer(Schema.STRING)
       .topic("test-topic-1")
       .enableBatching(false)
       .create();
    
   producer.newMessage()
       .property("k1","7")
       .property("k2", "vvvv")
       .property("k3", "true")
       .value("hi, this msg from `pulsar-msg-filter-plugin`")
       .send();

2. 消费方使用admin配置订阅消费组过滤表达式，其key固定为 pulsar-msg-filter-expression

   注：复杂表达式记得添加 "" 防止被转义

   pulsar-admin topics update-subscription-properties --property --property pulsar-msg-filter-expression="double(k1)<6 || (k2=='vvvv' && k3=='true')" --subscription 订阅组名称 主题
   
   pulsar-admin topics get-subscription-properties --subscription 订阅组名称 主题

   如上配置修改后立马生效，无需在创建Consumer时再设置subscriptionProperties

     Consumer consumer = client.newConsumer()
       .subscriptionName("订阅组名称")
       .topic("主题")
       .subscribe();

   说明: pulsar-msg-filter-plugin插件（服务端）依赖消息的MessageMetadata，故需关闭发送端的batch操作，否则无效（.enableBatching(false)），如无法关闭可配合pulsar-msg-filter-interceptor 一起使用。

添加 pulsar-msg-filter-interceptor 依赖

 <dependency>
     <groupId>io.github.yangl</groupId>
     <artifactId>pulsar-msg-filter-interceptor</artifactId>
     <version>VERSION</version>
 </dependency>

创建Consumer实例时配置 MsgFilterConsumerInterceptor 过滤器

Consumer<String> consumer = client.newConsumer(Schema.STRING)
        .subscriptionName("订阅组名称")
        .topic("主题")
        .intercept(MsgFilterConsumerInterceptor.<String>builder().build())
        .subscribe();

.intercept(MsgFilterConsumerInterceptor.<String>builder().webServiceUrl(YOUR_HTTP_SERVICE_URL).build())