Kafka 单机和分布式环境搭建与案例使用

旧城等待， 2022-06-10 10:47 262阅读 0赞

> 实验环境：  
> 1、Ubuntu Server 16.04  
> 2、kafka\_2.11-0.11.0.0

## 一、单机环境搭建 ##

官方参考文章：

[http://kafka.apache.org/quickstart][http_kafka.apache.org_quickstart]

**1、下载和解压安装包**

这里下载了zookeeper和kafaka两个安装包，下载地址：

zookeeper：[http://www.apache.org/dyn/closer.cgi/zookeeper/][http_www.apache.org_dyn_closer.cgi_zookeeper]

kafka：[http://kafka.apache.org/downloads][http_kafka.apache.org_downloads]

![这里写图片描述][20170817104835706]

**2、启动Zookeeper服务**

这里的kafka默认是由内置的zookeeper的，如果使用内置的zookeeper的话，启动的方式如下：

![这里写图片描述][20170817105055154]

zookeeper的配置文件是在：/kafka\_2.12-0.11.0.0/config 目录下

![这里写图片描述][20170817105259790]

启动Zookeeper：

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

当看到如下信息的时候，就表示成功了！

![这里写图片描述][20170817105449944]

**3、启动Kafka**

kafka的配置文件是在/kafka\_2.12-0.11.0.0/config 目录下，默认情况下不需要修改。

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

**4、创建一个Topic**

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

–replication-factor 复制因子为1；  
–partitions 分区为1；

查看已创建的Topic：

![这里写图片描述][20170817111946864]

**5、发送测试消息**

kafka支持从Console发送信息，消费者从Console接受信息。

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

–broker-list 表示代理服务器的列表，这里只有一个；

创建一个消费者：

>bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic testTopic --from-beginning

–from-beginning 表示从消息开始处读取；

然后在生产者的Console输入数据，消费者的Console就可以看到信息：

![这里写图片描述][20170817112949427]

![这里写图片描述][20170817113006071]

## 二、伪集群环境搭建 ##

官方提供了一种方式在一台机器上启动多个Broker机器构成multi-broker cluster，这是一种伪集群的方式，下边就配置一下。

1、修改配置文件

思路是配置多个config/server.properties文件，修改其中的`broker.id=1` 和端口号，日志文件位置。

> cp config/server.properties config/server-1.properties
    > cp config/server.properties config/server-2.properties

编辑配置文件，修改如下对应的位置：

config/server-1.properties:
        broker.id=1
        listeners=PLAINTEXT://:9093
        log.dir=/tmp/kafka-logs-1
     
    config/server-2.properties:
        broker.id=2
        listeners=PLAINTEXT://:9094
        log.dir=/tmp/kafka-logs-2

2、分别启动另外两个Kafka

>bin/kafka-server-start.sh config/server-1.properties &
    >>bin/kafka-server-start.sh config/server-2.properties &

&表示在后台运行

3、查看运行结果：

![这里写图片描述][20170817120028145]

QuorumPeerMain表示Zookeeper进行；  
另外有3个Kafka进程；

4、创建Topic

新建一个复制因子为3的Topic

>bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 3 --partitions 1 --topic my-replicated-topic

![这里写图片描述][20170817125136445]

查看Topic的描述信息：

>bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic my-replicated-topic

![这里写图片描述][20170817125720901]

5、发送消息

启动生产者，这里有3个Kafka实例，但是–broker-list 仍是启动的Zookeeper服务。

>bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-replicated-topic

启动消费者：

>bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic my-replicated-topic

和单机的情况是一样的。

## 三、分布式集群环境搭建 ##

搭建的分布式集群和伪集群的方式大致相同，这里假设使用3台服务器模拟实验，部署3个Zookeeper实例和3个Kafka实例，当然也可以直接部署一个Zookeeper实例，这里只是演示分布式Zookeeper和kafka的搭建。

![这里写图片描述][20170817132001786]

工具使用的是SecureCRT。

**1、分布式Zookeeper的搭建**

（1）将Zookeeper安装包分别上传到3台服务器，我的是放在：`/home/xuliugen/server` 目录下。

（2）配置第一台Zookeeper

复制`zookeeper-3.4.10/conf/zoo_sample.cfg` 为 `zookeeper-3.4.10/conf/zoo.cfg`，修改zoo.cfg文件如下，只更改data的目录：

![这里写图片描述][20170817132926421]

因为，修改了dataDir目录的位置，那么就需要创建一个`/zookeeper-3.4.6/data`目录。

（3）按同样的方式修改第二台Zookeeper和第三台Zookeeper服务器配置。

（4）然后，在每一台Zookeeper的配置文件中的最下边添加Zookeeper的集群配置：

![这里写图片描述][20170824103619697]

（5）最后创建每一个Zookeeper的 `myid` 文件，在`/data/myid`文件

xuliugen@xuliugen-pc:~/server/zookeeper-3.4.6/data$ echo 1 > myid

则，另外两台分别为：

xuliugen@xuliugen-pc:~/server/zookeeper-3.4.6/data$ echo 2 > myid

xuliugen@xuliugen-pc:~/server/zookeeper-3.4.6/data$ echo 3 > myid

**注意：**

1、myid和IP地址的对应

server.1=
    server.2=
    server.3=

这里的1、2、3是和我们刚才配置的myid的数值是相对应的，即1的IP地址为192.168.1.120，那么server.1=192.168.1.120:2888:3888

2、防火墙端口的配置

另外，2888:3888端口要设置防火墙权限

**2、启动Zookeeper服务器**

依此使用命令`./bin/zkServer.sh start` 启动Zookeeper服务。

使用`jps` 查看是否已经启动

![这里写图片描述][20170817133418936]

查看zookeeper日志的话，是在`/zookeeper-3.4.6/bin` 目录下的`zookeeper.out` 文件：

![这里写图片描述][20170824112931543]

使用`tailf zookeeper.out` 可以进行查看。

**3、分布式Kafka的搭建**

（1）将Kafka安装包分别上传到3台服务器，我的是放在：`/home/xuliugen/server` 目录下。

（2）配置第一台Kafka

Kafka的配置文件是在`/conf/server.properties` ，修改日志的目录：

![这里写图片描述][20170817134822284]

配置主机IP或者hostname：

![这里写图片描述][20170817165428453]

然后修改kafka中使用的Zookeeper集群地址：

![这里写图片描述][20170817134039511]

多个Zookeeper之间以英文逗号分开。

**注意：**

这里需要注意的是，如果按照上述的方式配置：

listeners=PLAINTEXT://192.168.1.120:9092

这样配置的话，是在内网环境下允许的，如果使用外网进行访问的话，可以配置为如下:

![这里写图片描述][20170913214108650]

具体请参考：  
[http://blog.csdn.net/fengcai19/article/details/54695874?utm\_source=itdadao&utm\_medium=referral][http_blog.csdn.net_fengcai19_article_details_54695874_utm_source_itdadao_utm_medium_referral]

（3）按同样的方式配置第二台kafka和第三台kafka服务器。

要注意的是不同的kafka的`broker.id` 一定要不一样，我这里分别配置的是0、1、2。

![这里写图片描述][20170817134515583]

**4、分别启动Kafka服务**

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

## 四、代码测试 ##

1、项目结构

![这里写图片描述][20170817171127576]

2、pom文件内容

<dependencies>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka-clients</artifactId>
                <version>0.11.0.0</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-log4j12</artifactId>
                <version>1.7.12</version>
            </dependency>
            <dependency>
                <groupId>org.slf4j</groupId>
                <artifactId>slf4j-api</artifactId>
                <version>1.7.12</version>
            </dependency>
        </dependencies>

3、日志配置log4j.properties

log4j.rootLogger=DEBUG,rolling,errlog,stdout
    #stdout log
    log4j.appender.stdout=org.apache.log4j.ConsoleAppender
    log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
    log4j.appender.stdout.layout.ConversionPattern=%d{HH:mm:ss} [%-5p] %c{1}.%M:%L-[%X{traceId}]-%m%n
    #common log
    log4j.appender.rolling=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.rolling.File=${catalina.base}/logs/kafka-demo.log
    log4j.appender.rolling.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.rolling.layout=org.apache.log4j.PatternLayout
    log4j.appender.rolling.layout.ConversionPattern=%d{HH:mm:ss.SSS} [%-5p] %-20c{1} [%t]%x [%X{traceId}]-%m%n
    #error log
    log4j.appender.errlog=org.apache.log4j.DailyRollingFileAppender
    log4j.appender.errlog.Threshold=ERROR
    log4j.appender.errlog.File=${catalina.base}/logs/error.log
    log4j.appender.errlog.DatePattern='.'yyyy-MM-dd-HH
    log4j.appender.errlog.layout=org.apache.log4j.PatternLayout
    log4j.appender.errlog.layout.ConversionPattern=%d{MM-dd HH:mm:ss.SSS} [%-5p] %-20c{1} [%.11t] [%X{traceId}]%x-%m%n

3、生产者

import org.apache.kafka.clients.producer.KafkaProducer;
    import org.apache.kafka.clients.producer.ProducerRecord;
    
    import java.util.Properties;
    
    public class ProducerDemo {
    
        // Topic
        private static final String topic = "kafkaTopic";
    
        public static void main(String[] args) throws Exception {
    
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.1.120:9092,192.168.1.135:9093,192.168.1.227:9094");
            props.put("acks", "0");
            props.put("group.id", "1111");
            props.put("retries", "0");
            props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
            props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
    
            //生产者实例
            KafkaProducer producer = new KafkaProducer(props);
    
            int i = 1;
    
            // 发送业务消息
            // 读取文件 读取内存数据库 读socket端口
            while (true) {
                Thread.sleep(1000);
                producer.send(new ProducerRecord<String, String>(topic, "key:" + i, "value:" + i));
                System.out.println("key:" + i + " " + "value:" + i);
                i++;
            }
        }
    }

4、消费者

import org.apache.kafka.clients.consumer.ConsumerRecord;
    import org.apache.kafka.clients.consumer.ConsumerRecords;
    import org.apache.kafka.clients.consumer.KafkaConsumer;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import java.util.Arrays;
    import java.util.Properties;
    
    public class ConsumerDemo {
    
        private static final Logger logger = LoggerFactory.getLogger(ConsumerDemo.class);
        private static final String topic = "kafkaTopic";
    
        public static void main(String[] args) {
    
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.1.120:9092,192.168.1.135:9093,192.168.1.227:9094");
            props.put("group.id", "1111");
            props.put("enable.auto.commit", "true");
            props.put("auto.commit.interval.ms", "1000");
            props.put("auto.offset.reset", "earliest");
            props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    
            KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(props);
    
            consumer.subscribe(Arrays.asList(topic));
    
            while (true) {
                ConsumerRecords<String, String> records = consumer.poll(1000);
                for (ConsumerRecord<String, String> record : records) {
                    System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
                }
            }
        }
    }

5、测试结果

生产者：

![这里写图片描述][20170817171335993]

消费者：

![这里写图片描述][20170817171343271]

代码下载地址：[http://download.csdn.net/download/u010870518/9938605][http_download.csdn.net_download_u010870518_9938605]

也可以到官网下载Kafka的源代码包，包里边有example代码可以参考

[https://www.apache.org/dyn/closer.cgi?path=/kafka/0.11.0.0/kafka-0.11.0.0-src.tgz][https_www.apache.org_dyn_closer.cgi_path_kafka_0.11.0.0_kafka-0.11.0.0-src.tgz]

![这里写图片描述][20170817194537834]

[http_kafka.apache.org_quickstart]: http://kafka.apache.org/quickstart
[http_www.apache.org_dyn_closer.cgi_zookeeper]: http://www.apache.org/dyn/closer.cgi/zookeeper/
[http_kafka.apache.org_downloads]: http://kafka.apache.org/downloads
[20170817104835706]: /images/20220610/5bd12bb8b1634499845952c0f48eaff8.png
[20170817105055154]: /images/20220610/3ec108a877174ca2ba2f4f6bcc67b7d9.png
[20170817105259790]: /images/20220610/12ca8a13d1534c4fa3ba1af7bd21be4e.png
[20170817105449944]: /images/20220610/e57562248a564739aab34430855cb4ef.png
[20170817111946864]: /images/20220610/024bf0e72eec4dbf8b701531a21e6ead.png
[20170817112949427]: /images/20220610/5d15f3d97fa84f3cafccfe6db9c74758.png
[20170817113006071]: /images/20220610/75d27c23561d47cd8974317ffa7f0028.png
[20170817120028145]: /images/20220610/efb0f53e254e4bb79bdc0ccf362a55cc.png
[20170817125136445]: /images/20220610/8c8f651f1b434b0299feff49f1d90b21.png
[20170817125720901]: /images/20220610/43ce8963bc93499d8d4ac5fc1545d35a.png
[20170817132001786]: /images/20220610/742800f29a1d462bbd01fd6c9073798d.png
[20170817132926421]: /images/20220610/61af3a1fcb394f5796b7397c5c2b1865.png
[20170824103619697]: /images/20220610/62b11d95be184a1fac1bee6ca4912ae6.png
[20170817133418936]: /images/20220610/7921504a302045d0b5ed4380f6a227fa.png
[20170824112931543]: /images/20220610/e7c9a72476084d80a20baefa9318591d.png
[20170817134822284]: /images/20220610/6525a53895364282a419432b8bacbb32.png
[20170817165428453]: /images/20220610/f7cdd62f71204980a3f5712dc0307244.png
[20170817134039511]: /images/20220610/34bcc141ebf84e519d2412d8e4794287.png
[20170913214108650]: /images/20220610/93532e81e7774cf5a258fa79437cde92.png
[http_blog.csdn.net_fengcai19_article_details_54695874_utm_source_itdadao_utm_medium_referral]: http://blog.csdn.net/fengcai19/article/details/54695874?utm_source=itdadao&utm_medium=referral
[20170817134515583]: /images/20220610/0b0ad754c3db4a95aa792f06d4233ced.png
[20170817171127576]: /images/20220610/ed8f8c38c8844c68ac5282448cb98f6a.png
[20170817171335993]: /images/20220610/cc697b284d1841a5959447ea9d668224.png
[20170817171343271]: /images/20220610/2a0df9012ae04784a5f65816bfa23c5f.png
[http_download.csdn.net_download_u010870518_9938605]: http://download.csdn.net/download/u010870518/9938605
[https_www.apache.org_dyn_closer.cgi_path_kafka_0.11.0.0_kafka-0.11.0.0-src.tgz]: https://www.apache.org/dyn/closer.cgi?path=/kafka/0.11.0.0/kafka-0.11.0.0-src.tgz
[20170817194537834]: /images/20220610/dde61df5504546b4a6caf0780d9a1e3b.png