当前位置:首页 > 服务端 > Kafka的基本概念与安装指南(单机+集群同步)

Kafka的基本概念与安装指南(单机+集群同步)

2022年09月16日 19:22:09服务端4

最近在搞spark streaming,很自然的前端对接的就是kafka。不过在kafka的使用中还是遇到一些问题,比如mirrormaker莫名其妙的丢失数据[原因稍后再说],消费数据offset错乱[之后介绍spark streaming的时候再解释]

总之,还是遇到了不少的问题。本篇就从下面几个方面介绍一下kafka:

  1. 基本介绍
  2. 安装与helloworld
  3. producer
  4. consumer
  5. mirror maker跨集群同步
  6. 控制台

基本介绍

Kafka是一款分布式的消息队列框架,它由三个重要的部分组成:

  1. Producer 消息的生产者,负责生产消息
  2. Broker 消息的存储,负责消息的持久化与高可用
  3. Consumer 消息的消费者,负责消费消息

大致的结构如下:

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

消息则是通过topic进行标识,每个topic可以有多个partition分区组成。每一个parition内部消息是按照顺序写入的,所有的partition加起来才是全部的数据,也就是说kafka并不能保证全局有序,只能保证在某一个partition内部是有序的。

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

消费者消费数据的时候是根据一个叫做offset的游标来记录消费的位置,可以通俗的把它理解成递增的id。

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

消费者可以由多个组成一个消费者组,同一个消费者组内的数据不会重复消费。不过消费者的数量跟partition的数量是有关系的,如果只有一个partition,那么即便是由10个消费者,同一时间也只能由一个消费者进行消费。

另外,broker是负责消息的持久化,前面提到过消息是通过partition组织在一起的,物理上则是通过一个log文件来记录。如果有一条消息写入,就会追加到log文件的末尾,当大小超过一定的阈值后,就新建一个log文件。如果log文件的修改时间超过一定的阈值,kafka还会清理掉该文件。

原理的东西就简单说这么多,下面来看看安装与体验吧!

安装与hello world

按照官方文档的步骤,是最快的入门方式:

下载安装包

官方下载地址下载安装包,并参照对应的版本的文档即可,下载后执行下面的命令:

> tar -xzf kafka_2.11-0.9.0.0.tgz
> cd kafka_2.11-0.9.0.0

启动zookeeper

如果方便的话,最好自己额外安装zookeeper,或者与其他的组建公用一个zk,否则单独为了kafka运行一个zk还是挺浪费资源的。

> bin/zookeeper-server-start.sh config/zookeeper.properties

最好不要随意修改zk的地址,2181是默认的端口号,如果修改,后面启动kafka会很麻烦,修改的地方会很多。

启动kafka-broker

bin/kafka-server-start.sh config/server.properties

创建主题并查看

bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
bin/kafka-topics.sh --list --zookeeper localhost:2181

启动producer

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

启动consumer

bin/kafka-console-consumer.sh --zookeeper localhost:2181 --topic test --from-beginning

至此,单机版的kafka就搭建完成了!如果要创建kafka的集群,可以直接

producer例子

import kafka.producer.KeyedMessage;
import kafka.javaapi.producer.Producer;
import kafka.producer.ProducerConfig;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaProducer {
    private static final String TOPIC = "test"; //kafka创建的topic
    private static final String CONTENT = "This is a single message"; //要发送的内容
    private static final String BROKER_LIST = "xxxx:9092"; //broker的地址和端口
    private static final String SERIALIZER_CLASS = "kafka.serializer.StringEncoder"; // 序列化类

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("serializer.class", SERIALIZER_CLASS);
        props.put("metadata.broker.list", BROKER_LIST);


        ProducerConfig config = new ProducerConfig(props);
        Producer<String, String> producer = new Producer<String, String>(config);

        //Send one message.
        KeyedMessage<String, String> message =
                new KeyedMessage<String, String>(TOPIC, CONTENT);
        producer.send(message);

        //Send multiple messages.
        List<KeyedMessage<String,String>> messages =
                new ArrayList<KeyedMessage<String, String>>();
        for (int i = 0; i < 100; i++) {
            messages.add(new KeyedMessage<String, String>
                    (TOPIC, i+"Multiple message at a time. " + i));
        }
        producer.send(messages);
		producer.close();
    }
}

执行后,如果有一个consumer启动,就可以看到消息输出。

consumer例子

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

public class KafkaConsumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("zookeeper.connect", "xxxx:2181");
        props.put("group.id", "t1");

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put("xxx-topic", 1);
        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));
        Map<String, List<KafkaStream<byte[], byte[]>>> msgStreams = consumer.createMessageStreams(topicCountMap);
        List<KafkaStream<byte[], byte[]>> msgStreamList = msgStreams.get("test");

        for(KafkaStream stream : msgStreamList){
            ConsumerIterator<byte[], byte[]> iterator = stream.iterator();
            while(iterator.hasNext()) {
                String message = new String(iterator.next().message());
                if(message.contains("xxxx")){
                    System.out.println(message);
                }
            }
        }
    }
}

跨集群同步——mirror maker

如果公司有云环境,可能还涉及到多个集群环境数据的同步。那么官方提供了一个mirrormaker的工具,它其实就是封装了一个consumer和一个producer,把一个集群的数据,直接消费到另一个集群。

代码可以参考github:
https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/tools/MirrorMaker.scala

文档可以参考:
https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=27846330

我这里介绍一下它的用法,首先启动的脚本,官方已经封装到kafka解压后的bin目录下。

主要用到了kafka-run-class.sh,kafka-mirror-maker.sh脚本其实就是对它的一层封装:

exec $(dirname $0)/kafka-run-class.sh kafka.tools.MirrorMaker $@

然后需要创建两个配置文件,分别是consumer的配置文件和producer的配置文件:

consumer.properties

zookeeper.connect=xxxx:2181 
group.id=test-mirror

zookeeper.connect是想要消费的集群的zk地址,group.id是消费者组的id,一定别跟其他的mirrormaker搞到一起哈![这就是我开篇遇到的问题原因]。

producer.properties

zk.connect=localhost:2181
bootstrap.servers=localhost:9092

zk.connect是消息即将存储的zk地址, bootstrap.servers是消息即将存储的broker地址。(我试过没有bootstrap.servers的话,会报错)

然后执行下面的命令,启动脚本即可:

./kafka-run-class.sh kafka.tools.MirrorMaker --consumerrties --producer.config producer.properties --whitelist test --num.streams 2

num.streams控制了消费者的个数,必须要设置的。

这样就开启了mirrormaker服务,可以看到第一个集群的所有消息,都同步到了第二个集群。

控制台主要功能介绍

控制台可以安装kafka-manager进行监控与管理,安装的教程可以参考:
http://blog.csdn.net/lsshlsw/article/details/47300145

集群概况

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

主题

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

broker

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

消费者

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

Kafka的基本概念与安装指南(单机+集群同步) _ JavaClub全栈架构师技术笔记

参考

作者:xingoo
来源链接:https://www.cnblogs.com/xing901022/p/7994339.html

版权声明:
1、JavaClub(https://www.javaclub.cn)以学习交流为目的,由作者投稿、网友推荐和小编整理收藏优秀的IT技术及相关内容,包括但不限于文字、图片、音频、视频、软件、程序等,其均来自互联网,本站不享有版权,版权归原作者所有。

2、本站提供的内容仅用于个人学习、研究或欣赏,以及其他非商业性或非盈利性用途,但同时应遵守著作权法及其他相关法律的规定,不得侵犯相关权利人及本网站的合法权利。
3、本网站内容原作者如不愿意在本网站刊登内容,请及时通知本站(javaclubcn@163.com),我们将第一时间核实后及时予以删除。


本文链接:https://www.javaclub.cn/server/42221.html

标签: Kafka
分享给朋友:

“Kafka的基本概念与安装指南(单机+集群同步)” 的相关文章

kafka消息中间件-快速学习

为什么需要消息队列   周末无聊刷着手机,某宝网APP突然蹦出来一条消息“为了回馈老客户,女朋友买一送一,活动仅限今天!”。买一送一还有这种好事,那我可不能错过!忍不住立马点了去。于是选了两个最新款,下单、支付一气呵成!满足的躺在床上,想着马上有女朋友了,竟然幸福的失眠了…...

kafka集群搭建

kafka集群搭建

本文将记录使用kafka镜像,分别在两种场景下搭建3节点集群:1.在一台机器上使用容器方式安装kafka集群;2.在三台机器上使用容器方式安装kafka集群。 此次使用的是wurstmeister的,下载量是比较大的。使用下面命令下载: docker pull wur...

kafka消息长度限制

更改为10M 客户端代码增加:max_request_size=10485760, 服务端配置:replica.fetch.max.bytes=10485760,message.max.bytes=10485760...

【kafka】安装部署kafka集群(kafka版本:kafka_2.12-2.3.0)

3.2.1 下载kafka并安装kafka_2.12-2.3.0.tgz tar -zxvf kafka_2.12-2.3.0.tgz 3.2.2 配置kafka集群 在config/server.properties中修改参数: [had...

Kafka 安装和简单使用

Kafka 安装和简单使用

文章目录 Kafka 安装和简单使用 kafka下载地址 windows 系统...

kafka的基本概念和工作流程分析

kafka的基本概念和工作流程分析

为什么需要消息队列   周末无聊刷着手机,某宝网APP突然蹦出来一条消息“为了回馈老客户,女朋友买一送一,活动仅限今天!”。买一送一还有这种好事,那我可不能错过!忍不住立马点了去。于是选了两个最新款,下单、支付一气呵成!满足的躺在床上,想着马上有女朋友了,竟然幸福的失眠了…...

Linux下Kafka下载与安装教程

Linux下Kafka下载与安装教程

原文链接:http://www.studyshare.cn/software/details/1176/0 一、预备环境 Kafka是java生态圈中的一员,运行在java虚拟机上,按Kafka官方说明,java环境推荐Java8;Kafka需要Zookeeper保存集群的...

Docker 安装 kafka

Docker 安装 kafka

简单安装为了集成 SpringBoot,真实使用,增加增加更多配置,比如将log映射出来 1.安装 zookeeper [root@centos-linux ~]# docker pull wurstmeister/zookeeper [root@centos-l...

Kafka如何保证消息不丢失不重复

首先需要思考下边几个问题: 消息丢失是什么造成的,从生产端和消费端两个角度来考虑 消息重复是什么造成的,从生产端和消费端两个角度来考虑 如何保证消息有序 如果保证消息不重不漏,损失的是什么 大概总结下 消费端重复消费:建立去重表 消费端丢失数据...

kafka查看topic列表和topic消息

kafka查看topic列表和topic消息

查询topic 列表信息 前提是需要进入到kafka的目录 Linux 目录 \kafka_2.12-2.8.0\bin\ sh kafka-topics.sh --list --zookeeper localhost:2181 windo...

发表评论

访客

◎欢迎参与讨论,请在这里发表您的看法和观点。