11_Pulsar Adaptors适配器、kafka适配器、Spark适配器

这篇具有很好参考价值的文章主要介绍了11_Pulsar Adaptors适配器、kafka适配器、Spark适配器。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

2.3. Pulsar Adaptors适配器
2.3.1.kafka适配器
2.3.2.Spark适配器

2.3. Pulsar Adaptors适配器

2.3.1.kafka适配器

Pulsar 为使用 Apache Kafka Java 客户端 API 编写的应用程序提供了一个简单的解决方案。
在生产者中, 如果想不改变原有kafka的代码架构, 就切换到Pulsar的平台中, 那么Pulsar adaptor on kafka就变的非常的有用了, 它可以帮助我们在不改变原有kafka的代码基础上, 即可接入pulsar, 但是需要注意, 相关配置信息需要进行一些调整, 例如: 地址与topic

  • 1- 需要导入Pulsar集成kafka的依赖包, 删除掉原有Kafka-client包
<dependency> 
     <groupId>org.apache.pulsar</groupId> 
     <artifactId>pulsar-client-kafka</artifactId> 
     <version>2.8.0</version> 
</dependency>

注: 目前Pulsar并在Maven中央仓库中并没有提供Pulsar-client-kafka 2.8.1的包, 故此处导入2.8.0

  • 2-编写生产者
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaAdaptorProducer {

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        //1. 创建kafka生产者的核心类对象: KafkaProducer
        // 1.1: 创建生产者配置对象: 设置相关配置
        Properties props = new Properties();
        props.put("bootstrap.servers", "pulsar://node1:6650,node2:6650,node3:6650");
        // 消息的确认方案
        props.put("acks", "all");
        // key序列化类型
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        // value 序列化类型
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        Producer<String, String> producer = new KafkaProducer<>(props); 
        //2. 发送数据 
        for (int i = 0; i < 10; i++) { 
            //2.1: 创建 生产者数据承载对象 一个对象代表是一条消息数据
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("persistent://public/default/txn_t1",Integer.toString(i), Integer.toString(i)); 
            producer.send(producerRecord).get(); 
        }
        
        //3. 释放资源 
        producer.close();
    }

}
  • 3-编写消费者
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

public class KafkaAdaptorConsumer {
    public static void main(String[] args) {
        //1. 创建kafka的消费者的核心对象: KafkaConsumer
        //1.1: 创建消费者配置对象, 并设置相关的参数:
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "pulsar://node1:6650,node2:6650,node3:6650");
        //消费者组的 id
        props.setProperty("group.id", "test");
        //是否启动消费者自动提交消费偏移量
        props.setProperty("enable.auto.commit", "true");
        //每间隔多长时间提交一次偏移量:单位 毫秒
        props.setProperty("auto.commit.interval.ms","1000");
        //key 反序列化
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //val 发序列化
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        //2. 给消费者设置订阅topic:
        consumer.subscribe(Arrays.asList("persistent://public/default/txn_t1"));
        //3. 循环获取相关的消息数据
        while (true) {
            //3.1: 从kafka中获取消息数据: 参数表示等待超时时间
            //注意: 如果没有获取到数据, 返回一个空集合对象, 如果数据集合中有多个 ConsumerRecord 对象
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            //3.2 遍历ConsumerRecords 获取每一个 ConsumerRecord 对象 : ConsumerRecord 消费者数据承载对象, 一个对象就是一条消息
            for (ConsumerRecord<String, String> record : records) {
                String massage = record.value();
                System.out.println("消息数据为:"+massage);
            }
        } 
    } 
}
  • 4- 先运行消费者, 进行监听, 然后运行生产者, 观察消费者是否可以正常消费到数据
    11_Pulsar Adaptors适配器、kafka适配器、Spark适配器,# Apache Pulsar,pulsar

2.3.2.Spark适配器

Pulsar 的 Spark Streaming 接收器是一个自定义的接收器,它使用 Apache Spark Streaming 能够从 Pulsar 接
收原始数据。

应用程序可以通过 Spark Streaming receiver 接收 Resilient Distributed Dataset (RDD) 格式的数据,并可
以通过多种方式对其进行处理。文章来源地址https://www.toymoban.com/news/detail-634637.html

  • 1-导入相关的依赖包
<dependency>
    <groupId>org.apache.pulsar</groupId>
    <artifactId>pulsar-spark</artifactId>
    <version>2.8.0</version>
</dependency>
  • 2-编写spark的流式代码
String serviceUrl = "pulsar://localhost:6650/"; 
String topic = "persistent://public/default/test_src"; 
String subs = "test_sub"; 
SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("Pulsar Spark Example"); 
JavaStreamingContext jsc = new JavaStreamingContext(sparkConf, Durations.seconds(60)); 
ConsumerConfigurationData<byte[]> pulsarConf = new ConsumerConfigurationData(); 
Set<String> set = new HashSet<>(); 
set.add(topic); 
pulsarConf.setTopicNames(set); 
pulsarConf.setSubscriptionName(subs); 
SparkStreamingPulsarReceiver pulsarReceiver = new SparkStreamingPulsarReceiver( 
serviceUrl, 
pulsarConf, 
new AuthenticationDisabled()); 
JavaReceiverInputDStream<byte[]> lineDStream = jsc.receiverStream(pulsarReceiver);

到了这里,关于11_Pulsar Adaptors适配器、kafka适配器、Spark适配器的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • win11新电脑转win10后发现网络无法连接,打开网络适配器,网络连接是空的

           win11新电脑因为要编程,怕有影响现在使用的软件,就直接USB还原成win10。还原好,发现网络连不上了,插上网线也无法连接,然后去了网络适配器发现啥都没有,一开始我人都蒙了-。-头一次遇到。 具体图片如下:        根据百度上大佬,我试了很多办法,都

    2024年02月12日
    浏览(94)
  • 网络适配器是什么 网络适配器有什么用

    网络适配器是什么? 网络适配器又称网卡或网络接口卡(NIC),英文名NetworkInterfaceCard。它是使计算机联网的设备。平常所说的网卡就是将PC机和LAN连接的网络适配器。网卡(NIC) 插在计算机主板插槽中,负责将用户要传递的数据转换为网络上其它设备能够识别的格式,通过

    2024年02月05日
    浏览(57)
  • 适配器模式:代理、适配器、桥接、装饰,这四个模式有何区别?

            关于适配器模式,今天我们主要学习它的两种实现方式,类适配器和对象适配器,以及5种常见的应用场景。同时,我还会通过剖析slf4j日志框架,来给你展示这个模式在真实项目中的应用。除此之外,在文章的最后,我还对代理、桥接、装饰器、适配器,这4种代

    2024年02月13日
    浏览(56)
  • 【C++】STL 算法 ⑩ ( 函数适配器 | 函数适配器概念 | 函数适配器分类 | 函数适配器辅助函数 | std::bind2nd 函数原型及示例 | std::bind 函数原型及示例 )

    在 STL 中 预定义了很多 函数对象 , 如果要 对 函数对象 的 参数 / 返回值 进行 计算 或 设置 , 可以 使用 \\\" 函数适配器 \\\" 实现上述需求 ; \\\" 函数适配器 \\\" 可以 将 已存在的 函数对象 转化为 另一种符合要求的 函数对象 ; \\\" 函数适配器 \\\" 定义在 functional 头文件 中 ; \\\" 函数适配器

    2024年02月02日
    浏览(65)
  • 网络适配器没有启用tcp/ip服务,WLAN 适配器的驱动程序可能出现问题

    笔记本抽风。登得上wifi和热点,但网不能用,“无法访问Internet”   win10自带的网络诊断提示: “找到问题 WLAN 适配器的驱动程序可能出现问题 Windows 无法自动将 IP 协议堆栈绑定到网络适配器。 未修复 无线网络 适配器出现问题 已失败 ” 试了试火绒的断网修复,提示网络

    2024年02月11日
    浏览(54)
  • 适配器模式

    将一个类的接口转换成客户希望的另一个接口,使得原本由于接口不兼容而不能一起工作的那些类可以一起工作 想使用一个已经存在的类,而他的接口不符合你的需求 想创建一个可以复用的类,该类可以与其他不相关的类或不可预见的类协同工作——? 想使用一些已经存在

    2024年02月02日
    浏览(27)
  • [go] 适配器模式

    将一个类的接口,转换成客户期望的另一个接口。适配器让原来接口不兼容的类可以合作无间。 Client:是包含当前程序业务逻辑的类 客户端代码只需通过与适配器交互即可,无需与具体的适配器耦合。因此,你可以向程序中添加新类型的适配器而无需修改已有代码。这在服

    2024年01月19日
    浏览(40)
  • C++适配器模式

    1 简介: 适配器模式是一种结构型设计模式,用于将一个类的接口转换为客户端所期望的另一个接口。适配器模式允许不兼容的类能够协同工作,通过适配器类来实现接口的转换和适配。 2 实现步骤: 以下是使用C++实现适配器模式的步骤: a. 定义目标接口:首先,确定客户

    2024年02月12日
    浏览(38)
  • 适配器设计模式

    一、适配器模式 B站:java架构师 定义:适配器模式把一个类的接口变换成客户端所期待的另一种接口,从而使原本因接口不匹配而无法在一起工作的两个类能够在一起工作 三种适配器:类的适配器模式、对象的适配器模式、接口的适配器模式 1.类适配器模式 实现方式:让

    2024年02月11日
    浏览(52)
  • 九、适配器模式

    一、什么是适配器模式   适配器模式(Adapter)的定义如下:将一个类的接口转换成客户希望的另外一个接口,使得原本由于接口不兼容而不能一起工作的那些类能一起工作。    适配器模式(Adapter)包含以下主要角色: 目标(Target)接口:当前系统业务所期待的接口,

    2024年02月10日
    浏览(36)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包