简单实现,在nodejs中简单使用kafka

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

什么是 Kafka

Kafka 是由 Linkedin 公司开发的,它是一个分布式的,支持多分区、多副本,基于 Zookeeper 的分布式消息流平台,它同时也是一款开源的基于发布订阅模式的消息引擎系统

Kafka 的基本术语

消息:Kafka 中的数据单元被称为消息,也被称为记录,可以把它看作数据库表中某一行的记录。

批次:为了提高效率, 消息会分批次写入 Kafka,批次就代指的是一组消息。

主题:消息的种类称为 主题(Topic),可以说一个主题代表了一类消息。相当于是对消息进行分类。主题就像是数据库中的表。

分区:主题可以被分为若干个分区(partition),同一个主题中的分区可以不在一个机器上,有可能会部署在多个机器上,由此来实现 kafka 的伸缩性,单一主题中的分区有序,但是无法保证主题中所有的分区有序

一、本地简单搭建kafka

1.1 下载并解压2.8.1版本 或者其他版本(点击下载:Apache Kafka)

kafkajs,kafka,分布式,数据分析

下载后,解压到指定文件夹,并创建两个文件夹以后使用

kafkajs,kafka,分布式,数据分析

1.2修改(如下图)配置文件

1.config下的zookeeper.properties,修改为刚才创建data的真实路径

kafkajs,kafka,分布式,数据分析

2.config下的server.properties,修改为刚才创建kafka-log的真实路径

kafkajs,kafka,分布式,数据分析

1.3启动项目 顺序启动

1.启动zookeeper

到解压文件夹下执行:不要关闭窗口

.\bin\windows\zookeeper-server-start.bat .\config\zookeeper.properties

2.启动kafka

到解压文件夹下执行:不要关闭窗口

.\bin\windows\kafka-server-start.bat .\config\server.properties

3.创建一个为test的topic

不要关闭窗口

.\bin\windows\kafka-topics.bat --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test1

4.创建生产者

到解压文件夹下执行:

.\bin\windows\kafka-console-producer.bat --broker-list localhost:9092 --topic test1


5.创建消费者

.\bin\windows\kafka-console-consumer.bat --bootstrap-server localhost:9092 --topic test1 --from-beginning

6.测试一下

生产者发消息后消费者是否可以收到

kafkajs,kafka,分布式,数据分析

二.在nodejs中使用kafka

2.1安装kafka-node
命令 npm install kafka-node
kafkajs,kafka,分布式,数据分析
2.2 创建生产者
import kafka  = require('kafka-node');
var client = new kafka.KafkaClient({kafkaHost:'127.0.0.1:9092'});
var Producer = kafka.Producer;
var topic = 'test1';
var payloads = [{  // 需要发送的一些配置信息
        topic: topic,
        messages: 'arrange'  // 需要生产的消息
    }];  // 此处必须要使用数组的形式,因为payloads的遍历采用的是foreach
producer.on('ready', function () {
   producer.send(payloads, function (err, data) {
      console.log(payloads);
      console.log("=======");
      console.log(data);
   });
})

producer.on('error', function (err) {
  console.log('error', err);
});

2.3创建消费者

import kafka  = require('kafka-node');
var client = new kafka.KafkaClient({kafkaHost:'127.0.0.1:9092'});
var topic = 'test1';
var payloads = [{  // 需要发送的一些配置信息
        topic: topic,
        messages: 'arrange'  // 需要生产的消息
    }];  // 此处必须要使用数组的形式,因为payloads的遍历采用的是foreach
var options = {  // 消费者的选择
    host: 'localhost:9092',
    sessionTimeout: 15000,
    autoCommit: true
};
var consumer = new kafka.Consumer(client, payloads, options);
consumer.on('message', function (message) {
    console.log(message);
});
 文章来源地址https://www.toymoban.com/news/detail-770478.html

到了这里,关于简单实现,在nodejs中简单使用kafka的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

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

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

相关文章

  • docker+jmeter实现简单的分布式压力测试

    阶梯压力测试的场景是为了验证在系统运行期间,用户不断的登录系统并使用一段时间,通过阶梯的场景,模拟实际应用期间系统的稳定性、承载性。 二、常用的阶梯压力测试线程组 stepping Thread Group 步长插件 Concurrency Thread Group 步长并发插件 两者区别 stg不提供设置启动延迟

    2024年04月16日
    浏览(47)
  • 基于Mongodb分布式锁简单实现,解决定时任务并发执行问题

    我们日常开发过程,会有一些定时任务的代码来统计一些系统运行数据,但是我们应用有需要部署多个实例,传统的通过配置文件来控制定时任务是否启动又太过繁琐,而且还经常出错,导致一些异常数据的产生 网上有很多分布式锁的实现方案,基于redis、zk、等有很多,但

    2023年04月18日
    浏览(35)
  • 【Spring Boot+Kafka+Mysql+HBase】实现分布式优惠券后台应用系统(附源码)

    需要全部代码请点赞关注收藏后评论区留言私信~~~ 分布式优惠券后台应用系统服务于两类用户群体,一类是商户,商户可以根据自己的实际情况即进行优惠券投放,另一类是平台消费用户,用户可以去领取商户发放的优惠券 分布式优惠券后台应用系统采用SpringBoot作为主体开

    2024年02月10日
    浏览(29)
  • 分布式消息服务kafka

    什么是消息中间件? 消息中间件是分布式系统中重要的组件,本质就是一个具有接收消息、存储消息、分发消息的队列,应用程序通过读写队列消息来通信。 例如:在淘宝购物时,订单系统处理完订单后,把订单消息发送到消息中间件中,由消息中间件将订单消息分发到下

    2024年02月01日
    浏览(35)
  • 【分布式技术】消息队列Kafka

    目录 一、Kafka概述 二、消息队列Kafka的好处 三、消息队列Kafka的两种模式 四、Kafka 1、Kafka 定义 2、Kafka 简介 3、Kafka 的特性 五、Kafka的系统架构 六、实操部署Kafka集群  步骤一:在每一个zookeeper节点上完成kafka部署 ​编辑 步骤二:传给其他节点 步骤三:启动3个节点 kafka管理

    2024年01月23日
    浏览(42)
  • 使用redis实现分布式锁

    在一个分布式系统中,也会涉及多个节点访问同一个公共资源的情况,此时就需要通过锁来做互斥控制,避免出现类似于“线程安全”的问题,而java的synchronized这样的锁只能在当前进程中生效,在分布式的这种多个进程多个主机的场景无能为力,此时就需要分布式锁。 例如

    2024年02月07日
    浏览(35)
  • 使用ZooKeeper实现分布式锁

    目录 引言 1. ZooKeeper简介 2. 分布式锁实现原理 3. 分布式锁实现步骤 步骤一:创建ZooKeeper客户端 步骤二:创建分布式锁类 步骤三:使用分布式锁 4. 总结 在分布式系统中,实现分布式锁是一项常见的任务,可以用于保证同一时间只有一个客户端可以访问共享资源,从而避免竞

    2024年02月21日
    浏览(37)
  • 【分布式应用】kafka集群、Filebeat+Kafka+ELK搭建

    主要原因是由于在高并发环境下,同步请求来不及处理,请求往往会发生阻塞。比如大量的请求并发访问数据库,导致行锁表锁,最后请求线程会堆积过多,从而触发 too many connection 错误,引发雪崩效应。 我们使用消息队列,通过异步处理请求,从而缓解系统的压力。消息队

    2024年02月16日
    浏览(28)
  • 分布式 - 消息队列Kafka:Kafka 消费者的消费位移

    01. Kafka 分区位移 对于Kafka中的分区而言,它的每条消息都有唯一的offset,用来表示消息在分区中对应的位置。偏移量从0开始,每个新消息的偏移量比前一个消息的偏移量大1。 每条消息在分区中的位置信息由一个叫位移(Offset)的数据来表征。分区位移总是从 0 开始,假设一

    2024年02月12日
    浏览(35)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包