Pulsar Function 例子

在单机环境下实现字符串追加函数(Pulsar 2.4.2版本)

1 启动单机Pulsar

     $ bin/pulsar-daemon start standalone

2 创建函数

1) 准备环境

    项目引用 compile 'org.apache.pulsar:pulsar-functions-api:2.4.2'

2) 创建JAVA函数(此函数用于数据源来的topic schema是string,输出的tiopic schema是string)

     导出jar包,放到pulsar服务器目录下,本例子放在 /data/jar/下

3)使用命令行工具加载函数到Pulsar，                     

   bin/pulsar-admin functions create \

   --classname test.AppStrFunction \

   --jar /data/jar/pf.jar \

   --inputs persistent://public/default/tlstest \

   --output persistent://public/default/teststr \

   --tenant public \

   --namespace default \

   --name appStrFunction

   参数说明:

停止函数

bin/pulsar-admin functions stop \

--tenant public \

--namespace default \

--name appStrFunction

启动函数

bin/pulsar-admin functions start \

--tenant public \

--namespace default \

--name appStrFunction

删除函数

bin/pulsar-admin functions delete \

--tenant public \

--namespace default \

--name appStrFunction

函数的日志在 pulsar安装目录 /logs/functions下

3 测试函数

   根据前边函数已成功加载启动

1)向tlstest主题发送消息   

import java.util.concurrent.TimeUnit;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
public class SendMsgTest{
  public static void main(String[] args){
      String url="pulsar://192.168.1.48:6650";
  try{
     PulsarClient client =PulsarClient.builder()
           .serviceUrl(url)
           .connectionTimeout(10,TimeUnit.SECONDS)
           .build();
     Producer<String> producer=client.newProducer(Schema.STRING)
           .topic("tlstest")
           .sendTimeout(10,TimeUnit.SECONDS)
           .producerName("senduser")
           .create();
           producer.send("this is a book");
           System.out.print("send ok");
           client.close();
      }catch(Exception e){
        e.printStackTrace();
      }
  }
}

2)读取teststr主题消息

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionInitialPosition;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.impl.schema.JSONSchema;
import schema.OrderModel;
import com.alibaba.fastjson.JSON;
public class RecFunTest {
public static void main(String[] args) {
String url = "http://192.168.1.48:8080";
try{
  PulsarClient client =PulsarClient.builder()
    .serviceUrl(url)
    .build();
 Consumer<String> consumer=client.newConsumer(Schema.STRING)
    .topic("teststr")
    .subscriptionInitialPosition(SubscriptionInitialPosition.Earliest)
    .subscriptionType(SubscriptionType.Exclusive)//订阅模式  Exclusive(独占,默认模式) Failover（灾备）Shared（共享）
    .subscriptionName("wbq")//订阅者名称
    .subscribe();
 while (true) {
   Message<String> mondmsg = consumer.receive();
   String msg=mondmsg.getValue();
                System.out.println("receive message=:"+msg);
             }
  }catch(Exception e){
     e.printStackTrace();
  }
 }
}