Storm与Kafka的集成

快来打我* 2023-01-02 04:29 138阅读 0赞

### 目录 ###

*   *  一、深入Kafka服务架构原理
     *  二、kafka集群安装与部署
     *  三、基于maven构建Storm-kafka工程
     *  四、Storm-kafka编程模型

版本选择：

*  storm-1.2.3
 *  kafka\_2.11-0.8.2.2

## 一、深入Kafka服务架构原理 ##

行业里最流行的Message Queue产品； MQ的特点：生产者，消费者

Kafka是LinkedIn开源出来的一个高吞吐的分布式消息系统。 使用scala开发，支持多语言客户端（c++、java、python等）

其具有以下特点：

1.  支持高Throughput（高吞吐量）的应用。多分区
2.  无需停机即可扩展机器
3.  持久化：通过将数据持久化到硬盘以及replication防止数据丢失
4.  支持online和offline消费数据的场景。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]

*  Broker：消息中间件处理结点， 一个kafka节点就是一个broker
 *  Producer：消息发布者
 *  Consumer：消息订阅者

**kafka的消息分几个层次：**

1.  **Topic**：一类消息，例如page view日志，click日志等都可以以topic的形式存在，kafka集群能够同时负责 多个topic的分发
2.  **Partition**： Topic物理上的分组，一个topic可以分为多个partition(默认是2个)，每个partition是一个有序 的队列。partition中的每条消息都会被分配一个有序的id（offset，消费位置）。创建Topic时，指定 partition数，数据量大的就设置多一些partition。
3.  **Message**：消息，最小订阅单元

**消息数据处理流程：**

1.  Producer根据指定的partition方法（round-robin、hash等），将消息发布到指定topic的partition里 面
2.  kafka集群接收到Producer发过来的消息后，将其持久化到硬盘，并保留消息指定时长（可配置，默认7 天），而不关注消息是否被消费。
3.  Consumer从kafka集群消费（pull）数据，并控制获取消息的offset偏移量。**Pull是Consumer客户端主动拉取数据，Consumer自己记录好消费的offset位置。  
    Push是服务端主动推送数据给Consumer客户端，服务端压力大，推送的数据延迟较高。**

**Kafka如何实现高吞吐量：**  
High Throughput是kafka需要实现的核心目标之一，为此kafka做了以下一些设计：

（1）数据磁盘持久化：消息不在内存中cache，直接写入到磁盘，充分利用磁盘的顺序读写性能，所以 broker没有内存压力。

（2）zero-copy：减少IO操作步骤

（3）数据批量发送  
（4）数据压缩  
（5）Topic划分为多个partition，提高parallelism （并行度）

**Kafka如何实现负载均衡：**  
Kafka不是master/slave 架构，不存在单点故障，所以不需要所谓的HA（热备）。

1.  producer根据用户指定的算法，将消息发送到指定的partition；
2.  存在多个partiiton，每个partition有自己的replica（副本），每个replica分布在不同的Broker节 点上；
3.  多个partition时需要选取出leader partition（通过zk的选举机制），leader partition负责控制读 写，并由zookeeper负责fail over；
4.  通过zookeeper管理broker与consumer的动态加入与离开；

**扩容：**  
当需要增加broker节点时，新增的broker会向zookeeper注册，而producer及consumer会根据注册 在zookeeper上的watcher感知这些变化，并及时作出调整

## 二、kafka集群安装与部署 ##

这里的storm使用的是1.2.3，由于官网有指定要求，storm1.2.3对应的kafka的版本是0.8 。  
![在这里插入图片描述][20210104204956182.png_pic_center]  
官方版本：[https://archive.apache.org/dist][https_archive.apache.org_dist]

1.  配置Server.propertis文件
2.  分发到其他机器上
3.  修改其他机器上server.properties文件

详细安装kafka教程看这篇博客：  
[https://blog.csdn.net/weixin\_45366499/article/details/106952008][https_blog.csdn.net_weixin_45366499_article_details_106952008]

## 三、基于maven构建Storm-kafka工程 ##

配置pom.xml文件：

<dependency>
        <groupId>org.apache.storm</groupId>
        <artifactId>storm-kafka</artifactId>
        <version>1.2.3</version>
    </dependency>
    
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.11</artifactId>
        <version>0.8.2.2</version>
        <exclusions>
            <exclusion>
                <artifactId>zookeeper</artifactId>
                <groupId>org.apache.zookeeper</groupId>
            </exclusion>
    
            <exclusion>
                <artifactId>log4j</artifactId>
                <groupId>log4j</groupId>
            </exclusion>
    
            <exclusion>
                <artifactId>slf4j-log4j12</artifactId>
                <groupId>org.slf4j</groupId>
            </exclusion>
    
            <exclusion>
                <artifactId>scala-library</artifactId>
                <groupId>org.scala-lang</groupId>
            </exclusion>
            <exclusion>
                <artifactId>jopt-simple</artifactId>
                <groupId>net.sf.jopt-simple</groupId>
            </exclusion>
            <exclusion>
                <artifactId>zkclient</artifactId>
                <groupId>com.101tec</groupId>
            </exclusion>
        </exclusions>
    </dependency>
    
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>0.8.2.2</version>
    
        <exclusions>
            <exclusion>
                <artifactId>slf4j-log4j12</artifactId>
                <groupId>org.slf4j</groupId>
            </exclusion>
    
            <exclusion>
                <artifactId>lz4</artifactId>
                <groupId>net.jpountz.lz4</groupId>
            </exclusion>
        </exclusions>
    </dependency>
    <dependency>
        <groupId>org.scala-lang</groupId>
        <artifactId>scala-library</artifactId>
        <version>2.11.11</version>
    </dependency>

## 四、Storm-kafka编程模型 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]

--------------------

以上内容仅供参考学习，如有侵权请联系我删除！  
如果这篇文章对您有帮助，左下角的大拇指就是对博主最大的鼓励。  
您的鼓励就是博主最大的动力！

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/3fa40a1fd20f43759b5db5853bd0f190.png
[20210104204956182.png_pic_center]: /images/20221120/f9963043506e4cb3a4b0ac86da9a101f.png
[https_archive.apache.org_dist]: https://archive.apache.org/dist
[https_blog.csdn.net_weixin_45366499_article_details_106952008]: https://blog.csdn.net/weixin_45366499/article/details/106952008
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80NTM2NjQ5OQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/adc712be37a14b4d982b0128501349ad.png

Storm与Kafka的集成

发表评论取消回复

还没有评论，来说两句吧...

相关阅读