一、依赖
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>pulsar-demo</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><pulsar.version>3.2.0</pulsar.version></properties><!-- in your <properties> block --><!-- in your <dependencies> block --><dependencies><dependency><groupId>org.apache.pulsar</groupId><artifactId>pulsar-client-all</artifactId><version>${pulsar.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>1.13.6</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-java_2.11</artifactId><version>1.13.6</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_2.11</artifactId><version>1.13.6</version></dependency><!-- https://mvnrepository.com/artifact/io.streamnative.connectors/pulsar-flink-connector --><dependency><groupId>io.streamnative.connectors</groupId><artifactId>pulsar-flink-connector_2.11</artifactId><version>1.13.6.1-rc13</version><exclusions><exclusion><groupId>org.apache.pulsar</groupId><artifactId>pulsar-client-all</artifactId></exclusion></exclusions></dependency></dependencies></project>
二、demo程序
package cn.edu.tju.test2;import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.pulsar.FlinkPulsarSource;
import org.apache.flink.streaming.util.serialization.PulsarDeserializationSchema;import java.util.Properties;public class FlinkPulsar01 {private static String SERVER_4 = "xx.xx.xx.xx";public static void main(String[] args) throws Exception {//创建flink的流式执行环境StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();//配置pulsarString serviceUrl = "pulsar://" + SERVER_4 + ":6650";String adminUrl = "http://" + SERVER_4 + ":8080";Properties properties = new Properties();properties.put("topic", "persistent://public/default/my-topic");//创建flink sourceFlinkPulsarSource<String> source = new FlinkPulsarSource<String>(serviceUrl, adminUrl,PulsarDeserializationSchema.valueOnly(new SimpleStringSchema()), properties);//从topic 最早的数据开始消费source.setStartFromEarliest();DataStreamSource<String> dataStreamSource = env.addSource(source);//打印数据流dataStreamSource.print();//执行jobenv.execute("my job");}
}
)
