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模板网!

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

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

相关文章

  • Windows11系统的电脑中出现网络适配器VMnet1和VMnet8消失问题解决

            查看电脑中 高级网络设置  如下图;          如果这两个网卡消失,可能是自己误删了这两个网卡的原故(如下图), 也可能有其他原因,可以通过以下方式解决: 1.点击 Win窗口 ,在搜索框输入 控制面板 2.打开 控制面板   3.点击  卸载程序  出现(如下图),

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

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

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

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

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

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

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

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

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

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

    2024年02月11日
    浏览(49)
  • 设计模式——适配器

    说起适配器,大家第一个想到的可能就是电源适配器。 电源适配器的作用想必同学们也都清楚,那就是将220伏高电压转换成想要的5伏至20伏左右稳定的低电压。 从某种程度上讲,编程中经常提起的适配器模式的原理与上面讲到的基本是一致的。 用于将一个类的接口转换成另

    2024年02月12日
    浏览(52)
  • 适配器模式(C++)

    将一个类的接口转换成客户希望的另一个接口。Adapter模式使得原本由于接口不兼容而不能一起工作的那些类可以一起工作。 在软件系统中,由于应用环境的变化,常常需要将“一些现存的对象 ”放在新的环境中应用,但是新环境要求的接口是这些现存对象所不满足的。 如何

    2024年02月14日
    浏览(31)
  • 适配器模式介绍

    目录 一、适配器模式介绍 1.1 适配器模式定义 1.2 适配器模式原理 1.2.1 适配器模式类图 1.2.2 模式角色说明 二、适配器模式的应用 2.1 类适配器模式 2.1.1 需求说明 2.1.2 需求实现 2.1.2.1 类图 2.1.2.2 具体实现 2.1.2.2.1 SDCard接口 2.1.2.2.2 SDCardImpl实现类 2.1.2.2.3 TFCard接口 2.1.2.2.4 TFCard

    2024年01月17日
    浏览(37)
  • 适配器模式(Adapter)

    适配器是一种 结构型设计模式 ,它能使 接口不兼容的对象能够相互合作 。 封装器模式(Wrapper)。 1. 问题 假如你正在开发一款股票市场监测程序,它会从 不同来源下载 XML 格式的股票数据,然后向用户呈现出美观的图表 。 在开发过程中, 你决定在程序中 整合一个第三方

    2024年02月11日
    浏览(50)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包