Spring Boot 中使用 kafka

2017 年 12 月 1 日分類 database, hadoop

Kafka 是一种高吞吐的分布式发布订阅消息系统，能够替代传统的消息队列用于解耦合数据处理，缓存未处理消息等，同时具有更高的吞吐率，支持分区、多副本、冗余，因此被广泛用于大规模消息数据处理应用。Kafka 支持Java 及多种其它语言客户端，可与Hadoop、Storm、Spark等其它大数据工具结合使用。

环境安装

搭建高吞吐量 Kafka 分布式发布订阅消息集群

测试用例

Github 代码

代码我已放到 Github ，导入spring-boot-kafka 项目

github https://github.com/souyunku/spring-boot-examples/tree/master/spring-boot-kafka

添加依赖

在项目中添加 kafka-clients 依赖

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.10.2.0</version>
</dependency>
<dependency>
    <groupId>org.springframework.kafka</groupId>
    <artifactId>spring-kafka</artifactId>
</dependency>

启用 kafka

@Configuration
@EnableKafka
public class KafkaConfiguration {

}

消息生产者

@Component
public class MsgProducer {

    private static final Logger log = LoggerFactory.getLogger(MsgProducer.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void sendMessage(String topicName, String jsonData) {
        log.info("向kafka推送数据:[{}]", jsonData);
        try {
            kafkaTemplate.send(topicName, jsonData);
        } catch (Exception e) {
            log.error("发送数据出错！！！{}{}", topicName, jsonData);
            log.error("发送数据出错=====>", e);
        }

        //消息发送的监听器，用于回调返回信息
        kafkaTemplate.setProducerListener(new ProducerListener<String, String>() {
            @Override
            public void onSuccess(String topic, Integer partition, String key, String value, RecordMetadata recordMetadata) {
            }

            @Override
            public void onError(String topic, Integer partition, String key, String value, Exception exception) {
            }

            @Override
            public boolean isInterestedInSuccess() {
                log.info("数据发送完毕");
                return false;
            }
        });
    }

}

消息消费者

@Component
public class MsgConsumer {

    @KafkaListener(topics = {"topic-1","topic-2"})
    public void processMessage(String content) {

        System.out.println("消息被消费"+content);
    }
    
}

参数配置

application.properties

#kafka
# 指定kafka 代理地址，可以多个
spring.kafka.bootstrap-servers=YZ-PTEST-APP-HADOOP-02:9092,YZ-PTEST-APP-HADOOP-04:9092
# 指定listener 容器中的线程数，用于提高并发量
spring.kafka.listener.concurrency=3
# 每次批量发送消息的数量
spring.kafka.producer.batch-size=1000
# 指定默认消费者group id
spring.kafka.consumer.group-id=myGroup
# 指定默认topic id
spring.kafka.template.default-topic=topic-1

启动服务

@SpringBootApplication
@ComponentScan(value = {"io.ymq.kafka"})
public class Startup {

    public static void main(String[] args) {
        SpringApplication.run(Startup.class, args);
    }
}

单元测试

import io.ymq.kafka.MsgProducer;
import io.ymq.kafka.run.Startup;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;
/**
 * 描述: 测试 kafka
 *
 * @author yanpenglei
 * @create 2017-10-16 18:45
 **/
@RunWith(SpringRunner.class)
@SpringBootTest(classes = Startup.class)
public class BaseTest {

    @Autowired
    private MsgProducer msgProducer;

    @Test
    public void test() throws Exception {

        msgProducer.sendMessage("topic-1", "topic--------1");
        msgProducer.sendMessage("topic-2", "topic--------2");
    }
}

消息生产者,响应

2017-10-17 15:54:44.814  INFO 2960 --- [           main] io.ymq.kafka.MsgProducer                 : 向kafka推送数据:[topic--------1]
2017-10-17 15:54:44.860  INFO 2960 --- [           main] io.ymq.kafka.MsgProducer                 : 向kafka推送数据:[topic--------2]
2017-10-17 15:54:44.878  INFO 2960 --- [ad | producer-1] io.ymq.kafka.MsgProducer                 : 数据发送完毕
2017-10-17 15:54:44.878  INFO 2960 --- [ad | producer-1] io.ymq.kafka.MsgProducer                 : 数据发送完毕

消息消费者,响应

1 2	消息被消费topic--------1 消息被消费topic--------2

代码我已放到 Github ，导入spring-boot-kafka 项目

github https://github.com/souyunku/spring-boot-examples/tree/master/spring-boot-kafka

遇到一些坑

[2017-10-16 19:20:08.340] - 14884 严重 [main] --- org.springframework.kafka.support.LoggingProducerListener: Exception thrown when sending a message with key='null' and payload='topic--------2' to topic topic-2:

经调试发现 kafka 连接是用的主机名，所以修改 hosts

C:\Windows\System32\drivers\etc\hosts

10.32.32.149 YZ-PTEST-APP-HADOOP-02
10.32.32.154 YZ-PTEST-APP-HADOOP-04

留言與分享

搭建高吞吐量Kafka分布式发布订阅消息集群

2017 年 11 月 30 日分類 database, hadoop

搭建高吞吐量 Kafka 分布式发布订阅消息集群

简介

环境

Zookeeper集群: 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

kafka 集群: 192.168.252.124 , 192.168.252.125 , 192.168.252.126

kafka-manager: 192.168.252.127

主机名修改

CentOs7.3 修改主机名

ssh 免密登录

CentOs7.3 ssh 免密登录

安装 JDK1.8

CentOs7.3 安装 JDK1.8

搭建 Zookeeper 集群

CentOs7.3 搭建 ZooKeeper-3.4.9 Cluster 集群服务

Zookeeper集群: 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

主机名依次被我修改成: node1,node2,node3

搭建 kafka 集群

kafka 集群: 192.168.252.124 , 192.168.252.125 , 192.168.252.126

主机名依次被我修改成: node4,node5,node6

1.下载代码

kafka 官网下载 http://kafka.apache.org/downloads

阿里镜像:https://mirrors.aliyun.com/apache/kafka/

$ cd /opt
$ wget https://mirrors.tuna.tsinghua.edu.cn/apache/kafka/0.11.0.0/kafka_2.12-0.11.0.0.tgz
$ tar -zxvf kafka_2.12-0.11.0.0.tgz
$ cd kafka_2.12-0.11.0.0

2.修改配置

在 node4 操作

1	$ vi /opt/kafka_2.12-0.11.0.0/config/server.properties

broker.id=0  每台服务器不能重复

#设置zookeeper的集群地址
zookeeper.connect=192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

把配置复制到 node5,node6 集群

1	$ for a in {5..6} ; do scp -r /opt/kafka_2.12-0.11.0.0/ node$a:/opt/kafka_2.12-0.11.0.0 ; done

修改 node5,node6 集群的broker.id

1	$ vi /opt/kafka_2.12-0.11.0.0/config/server.properties

3.启动kafka

在 node1,启动 Kafka使用的 ZooKeeper，所以先启动ZooKeeper服务器

1	$ for a in {1..3} ; do ssh node$a "source /etc/profile; /opt/zookeeper-3.4.9/bin/zkServer.sh start" ; done

现在 node4 启动Kafka服务器

1	$ for a in {4..6} ; do ssh node$a "source /etc/profile; /opt/kafka_2.12-0.11.0.0/bin/kafka-server-start.sh /opt/kafka_2.12-0.11.0.0/config/server.properties" ; done

或者后台启动运行，日志查看去Kafka解压目录有个log 文件夹查看

1	$ for a in {4..6} ; do ssh node$a "source /etc/profile; nohup /opt/kafka_2.12-0.11.0.0/bin/kafka-server-start.sh /opt/kafka_2.12-0.11.0.0/config/server.properties > /dev/null 2>&1 &" ; done

查看进程，Kafka 是否启动成功

1
2
3

$ jps
3825 Kafka
6360 Jps

如果报错删除

1
2
3

kafka.common.KafkaException: Failed to acquire lock on file .lock in /tmp/kafka-logs. A Kafka instance in another process or thread is using this directory.

$ rm -rf /tmp/kafka-logs

4.创建主题

1	$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --create --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181 --replication-factor 2 --partitions 1 --topic ymq

--replication-factor 2 #复制两份
--partitions 1 #创建1个分区
--topic #主题为ymq

运行list topic命令，可以看到该主题：

1	$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --list --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181

5.生产消息

Kafka附带一个命令行客户端，它将从文件或标准输入中输入，并将其作为消息发送到Kafka群集。默认情况下，每行将作为单独的消息发送。

在 node5 运行生产者，然后在控制台中输入一些消息以发送到服务器。

1 2	$ /opt/kafka_2.12-0.11.0.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic ymq >www.ymq.io

6.消费消息

在node6 运行消费者，将把消息转储到标准输出。

1
2
3

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-console-consumer.sh --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181  --topic ymq --from-beginning
Using the ConsoleConsumer with old consumer is deprecated and will be removed in a future major release. Consider using the new consumer by passing [bootstrap-server] instead of [zookeeper]. 
www.ymq.io

7.topic详情

用describe 查看集群中topic每个节点情况

1
2
3

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --describe --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181 --topic ymq
Topic:ymq    PartitionCount:1    ReplicationFactor:2    Configs:
    Topic: ymq    Partition: 0    Leader: 1    Replicas: 1,3    Isr: 3,1

以下是输出的说明。第一行给出了所有分区的摘要，每个附加行提供有关一个分区的信息。由于我们这个主题只有一个分区，只有一行。

leader负责给定分区的读取和写入分配节点编号，每个分区的部分数据会随机指定不同的节点
replicas是复制此分区的日志的节点列表
isr一组正在同步的副本列表

8.删除topic

1
2
3

$ /opt/kafka_2.12-0.11.0.0/bin/kafka-topics.sh --delete --zookeeper 192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181 --topic ymq
Topic ymq is marked for deletion.
Note: This will have no impact if delete.topic.enable is not set to true.

9.停止kafka

1	$ for a in {4..6} ; do ssh node$a "source /etc/profile; /opt/kafka_2.12-0.11.0.0/bin/kafka-server-stop.sh /opt/kafka_2.12-0.11.0.0/config/server.properties " ; done

部署 Kafka Manager

Yahoo开源Kafka集群管理器Kafka Manager

作为一个分布式的消息发布-订阅系统，Apache Kafka在Yahoo内部已经被很多团队所使用，例如媒体分析团队就将其应用到了实时分析流水线中，同时，Yahoo整个Kafka集群处理的峰值带宽超过了20Gbps（压缩数据）。为了让开发者和服务工程师能够更加简单地维护Kafka集群，Yahoo构建了一个基于Web的管理工具，称为Kafka Manager，日前该项目已经在GitHub上开源。

通过Kafka Manager用户能够更容易地发现集群中哪些主题或者分区分布不均匀，同时能够管理多个集群，能够更容易地检查集群的状态，能够创建主题，执行首选的副本选择，能够基于集群当前的状态生成分区分配，并基于生成的分配执行分区的重分配，此外，Kafka Manager还是一个非常好的可以快速查看集群状态的工具。

Kafka Manager使用Scala语言编写，其Web控制台基于Play Framework实现，除此之外，Yahoo还迁移了一些Apache Kafka的帮助程序以便能够与Apache Curator框架一起工作。

一、它支持以下内容：

管理多个群集
容易检查集群状态（主题，消费者，偏移量，经纪人，副本分发，分区分配）
运行首选副本选举
使用选项生成分区分配，以选择要使用的代理
运行分区的重新分配（基于生成的分配）
创建可选主题配置的主题（0.8.1.1具有不同于0.8.2+的配置）
删除主题（仅支持0.8.2+，并记住在代理配置中设置delete.topic.enable = true）
主题列表现在表示标记为删除的主题（仅支持0.8.2+）
批量生成多个主题的分区分配，并选择要使用的代理
批量运行多个主题的分区重新分配
将分区添加到现有主题
更新现有主题的配置
可选地，启用JMX轮询代理级和主题级度量。
可选地筛选出在zookeeper中没有ids / owner /＆offset /目录的消费者。

源码，并编译打包

在 kafka-manager: 192.168.252.127 node7 部署

编译超级慢

$ yum install git
$ cd /opt/
$ git clone https://github.com/yahoo/kafka-manager
$ cd kafka-manager/
$ ./sbt clean dist

下载编译好的包

1
2
3

$ yum install unzip
$ unzip kafka-manager-1.3.2.1.zip
$ vi /opt/kafka-manager-1.3.2.1/conf/application.conf

修改这个 zk 地址

1	kafka-manager.zkhosts="192.168.252.121:2181,192.168.252.122:2181,192.168.252.123:2181"

启动 kafka-manager

默认端口 NettyServer - Listening for HTTP on /0:0:0:0:0:0:0:0:9000

1	$ /opt/kafka-manager-1.3.2.1/bin/kafka-manager -Dconfig.file=conf/application.conf

或者后台运行并且配置端口

1	$ nohup bin/kafka-manager -Dconfig.file=/home/hadoop/app/kafka-manager-1.3.2.1/conf/application.conf -Dhttp.port=9000 &

访问： http://ip:9000

图片描述

留言與分享

第 1 頁共 1 頁

Spring Boot 中使用 kafka

环境安装

测试用例

Github 代码

添加依赖

启用 kafka

消息生产者

消息消费者

参数配置

启动服务

单元测试

遇到一些坑

搭建高吞吐量Kafka分布式发布订阅消息集群

搭建高吞吐量 Kafka 分布式发布订阅消息 集群

简介

环境

主机名修改

ssh 免密登录

安装 JDK1.8

搭建 Zookeeper 集群

搭建 kafka 集群

1.下载代码

2.修改配置

3.启动kafka

4.创建主题

5.生产消息

6.消费消息

7.topic详情

8.删除topic

9.停止kafka

部署 Kafka Manager

源码，并编译打包

下载编译好的包

启动 kafka-manager

Kein Chan

R语言-环境安装

R语言-基础

R语言-读取数据

R语言-绘图

R语言-线性回归

1. 算法在计算中的作用

2. 算法基础

3. 函数的增长

4. 分治策略

5. 概率分析和随机算法

搭建高吞吐量 Kafka 分布式发布订阅消息集群