分类：Kafka

2022
05-07

SpringBoot+Nacos+Kafka微服务流编排的简单实现

目录前言准备工作Nacos安装及使用入门准备三个SpringBoot服务，引入Nacos及Kafka业务解读Nacos配置创建配置读取配置监听配置改变总结前言最近一直在做微服务开发，涉及了一些数据处理模块的开发，每个处理业务都会开发独立的微服务，便于后面拓展和流编排，学习了SpringCloudDataFlow等框架，感觉这个框架对于我们来说太重了，维护起来也比较麻烦，于是根据流编排的思想，基于我们目前的技术栈实现简单的流编排功能。简单的说，... 继续阅读 >

2022
02-11

使用spring boot 整合kafka,延迟启动消费者

springboot整合kafka,延迟启动消费者springboot整合kafka的时候一般使用@KafkaListener来设置消费者，但是这种方式在spring启动的时候就会立即开启消费者。如果有需要根据配置信息延迟开启指定的消费者就不能使用这种方式。参考了类：KafkaListenerAnnotationBeanPostProcessor，我提取了一部分代码。可以根据需要随时动态的开启消费者。还可以很方便的启动多个消费者。为了方便使用，我自定义了一个注解：importorg.springfra... 继续阅读 >

2021
10-12

Java Kafka 消费积压监控的示例代码

后端代码：Monitor.java代码：packagecom.suncreate.kafkaConsumerMonitor.service;importcom.suncreate.kafkaConsumerMonitor.model.ConsumerInfo;importorg.apache.kafka.clients.consumer.ConsumerConfig;importorg.apache.kafka.clients.consumer.KafkaConsumer;importorg.apache.kafka.clients.consumer.OffsetAndMetadata;importorg.apache.kafka.common.PartitionInfo;importorg.apache.kafka.common.TopicPartition... 继续阅读 >

2021
05-29

.Net Core 集成 Kafka的步骤

最近维护的一个系统并发有点高，所以想引入一个消息队列来进行削峰。考察了一些产品，最终决定使用kafka来当做消息队列。以下是关于kafka的一些知识的整理笔记。kafkakafka是分布式流式平台。它由linkedin开发，后贡献给了Apache开源组织并成为顶级开源项目。它可以应用在高并发场景下的日志系统，也可以当作消息队列来使用，也可以当作消息服务对系统进行解耦。流处理平台有以下三种特性:可以让你发布和订阅流式的记录。这一... 继续阅读 >

2021
04-27

Spring Boot 集成 Kafkad的实现示例

SpringBoot作为主流微服务框架，拥有成熟的社区生态。市场应用广泛，为了方便大家，整理了一个基于springboot的常用中间件快速集成入门系列手册，涉及RPC、缓存、消息队列、分库分表、注册中心、分布式配置等常用开源组件，大概有几十篇文章，陆续会开放出来，感兴趣同学请提前关注&收藏消息通信有两种基本模型，即发布-订阅（Pub-Sub）模型和点对点（PointtoPoint）模型，发布-订阅支持生产者消费者之间的一对多关系，而点对... 继续阅读 >

2021
01-08

SpringBoot集成Kafka的步骤

SpringBoot集成Kafka本篇主要讲解SpringBoot如何集成Kafka，并且简单的编写了一个Demo来测试发送和消费功能前言选择的版本如下：springboot：2.3.4.RELEASEspring-kafka：2.5.6.RELEASEkafka:2.5.1zookeeper:3.4.14本Demo使用的是SpringBoot比较高的版本SpringBoot2.3.4.RELEASE它会引入spring-kafka2.5.6RELEASE，对应了版本关系中的SpringBoot2.3usersshoulduse2.5.x(Bootdependencymanagement... 继续阅读 >

2020
11-11

Kafka producer端开发代码实例

一、producer工作流程producer使用用户启动producer的线程，将待发送的消息封装到一个ProducerRecord类实例，然后将其序列化之后发送给partitioner，再由后者确定目标分区后一同发送到位于producer程序中的一块内存缓冲区中。而producer的另外一个线程（Sender线程）则负责实时从该缓冲区中提取出准备就绪的消息封装进一个批次(batch)，统一发送给对应的broker，具体流程如下图：二、producer示例程序开发首先引入kafka相... 继续阅读 >

2020
11-11

Kafka多节点分布式集群搭建实现过程详解

上一篇分享了单节点伪分布式集群搭建方法，本篇来分享一下多节点分布式集群搭建方法。多节点分布式集群结构如下图所示：为了方便查阅，本篇将和上一篇一样从零开始一步一步进行集群搭建。一、安装Jdk具体安装步骤可参考linux安装jdk。二、安装与配置zookeeper下载地址：https://www-us.apache.org/dist/zookeeper/stable/下载二进制压缩包zookeeper-3.4.14.tar.gz，然后上传到linux服务器指定目录下，本次上传目录... 继续阅读 >

2020
11-11

Kafka单节点伪分布式集群搭建实现过程详解

Kafka集群搭建分为单节点的伪分布式集群和多节点的分布式集群两种，首先来看一下单节点伪分布式集群安装。单节点伪分布式集群是指集群由一台ZooKeeper服务器和一台Kafkabroker服务器组成，如下图所示：为了搭建单节点Kafka集群，需要依次安装如下软件：安装Java-->安装ZooKeeper-->安装Kafka。一、安装Java可以参考linux安装jdk,来进行安装二、安装ZooKeeper下载地址：https://www-us.apache.org/dist/zookeeper/st... 继续阅读 >

2020
10-30

Springboot 1.5.7整合Kafka-client代码示例

在一次项目中，因甲方需要使用kafka消息队列推送数据，所以需要接入kafka，并且kafka的版本是2.11。但是我们项目使用的是Springboot1.5.7的版本，对应的springboot.kafka.starter有冲突，所以就接入了kafka-client。Kafka是一个分布式消息引擎与流处理平台，经常用做企业的消息总线、实时数据管道，有的还把它当做存储系统来使用。早期Kafka的定位是一个高吞吐的分布式消息系统，目前则演变成了一个成熟的分布式消息引擎，以及... 继续阅读 >

2020
10-16

Spring boot 整合KAFKA消息队列的示例

这里使用spring-kafka依赖和KafkaTemplate对象来操作Kafka服务。一、添加依赖和添加配置项1.1、在Pom文件中添加依赖<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>1.2、添加配置项spring:kafka:bootstrap-servers:12.168.3.62:9092#指定kafka代理地址，可以多个producer:retries:2#写入失败时，重试次数。当retris为0时，produce不会重复... 继续阅读 >

2020
10-13

Python confluent kafka客户端配置kerberos认证流程详解

kafka的认证方式一般有如下3种：1.SASL/GSSAPI从版本0.9.0.0开始支持2.SASL/PLAIN从版本0.10.0.0开始支持3.SASL/SCRAM-SHA-256以及SASL/SCRAM-SHA-512从版本0.10.2.0开始支持其中第一种SASL/GSSAPI的认证就是kerberos认证，对于java来说有原生的支持，但是对于python来说配置稍微麻烦一些，下面说一下具体的配置过程，confluentkafka模块底层依赖于librdkafka，这是使用c编写的高性能的kafka客户端库，有好多语言的库都是依... 继续阅读 >

2020
10-10

.NET Core下使用Kafka的方法步骤

安装CentOS安装kafkaKafka:http://kafka.apache.org/downloadsZooLeeper:https://zookeeper.apache.org/releases.html下载并解压#下载，并解压$wgethttps://archive.apache.org/dist/kafka/2.1.1/kafka_2.12-2.1.1.tgz$tar-zxvfkafka_2.12-2.1.1.tgz$mvkafka_2.12-2.1.1.tgz/data/kafka#下载zookeeper，解压$wgethttps://mirror.bit.edu.cn/apache/zookeeper/zookeeper-3.5.8/apache-zookeeper-3.5.8-bin.tar.gz... 继续阅读 >

2020
10-07

Java kafka如何实现自定义分区类和拦截器

生产者发送到对应的分区有以下几种方式：（1）指定了patition，则直接使用；(可以查阅对应的javaapi,有多种参数)（2）未指定patition但指定key，通过对key的value进行hash出一个patition；（3）patition和key都未指定，使用轮询选出一个patition。但是kafka提供了，自定义分区算法的功能,由业务手动实现分布：1、实现一个自定义分区类，CustomPartitioner实现Partitionerimportorg.apache.kafka.clients.producer.Partitioner;i... 继续阅读 >

2020
10-06

Python通过kerberos安全认证操作kafka方式

如何通过Kerberos认证.1.安装Kerberos客户端CentOS：yuminstallkrb5-workstation使用whichkinit查看是否安装成功；2.拷贝Kerberos配置文件conf目录下krb5.conf和kafka.keytab和jaas.conf拷贝到客户端机器的etc目录,同时,krb5.conf中的kdc集群主机名和IP配置到客户端机器hosts配置文件中3.Kinit客户端通过kerberos认证获取Principalklist-ktkafka.keytab4.安装python-gssapipipinstallgssapi遇到的问题,如下:a.在linux中执... 继续阅读 >

2020
10-05

Kafka Java Producer代码实例详解

根据业务需要可以使用Kafka提供的JavaProducerAPI进行产生数据，并将产生的数据发送到Kafka对应Topic的对应分区中，入口类为：ProducerKafka的ProducerAPI主要提供下列三个方法：publicvoidsend(KeyedMessage<K,V>message)发送单条数据到Kafka集群publicvoidsend(List<KeyedMessage<K,V>>messages)发送多条数据(数据集)到Kafka集群publicvoidclose()关闭Kafka连接资源一、JavaKafkaProducerPartit... 继续阅读 >