Kafka 0.11.0.2 安装备忘录

悠悠 2022-06-05 09:20 70阅读 0赞

# Kafka 0.11.0.2 安装备忘录 #

#### 本文基于 Kafka 0.11.0.2的安装，环境 Centos 6 / Centos 7 ####

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

服务器共**3**台,用来安装Kafka集群和ZK集群

<table> 
 <thead> 
  <tr> 
   <th>服务器名</th> 
   <th>IP</th> 
   <th>zk myid</th> 
   <th>Kafka broker ID</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>zk1</td> 
   <td>10.20.32.113</td> 
   <td>1</td> 
   <td>1</td> 
  </tr> 
  <tr> 
   <td>zk2</td> 
   <td>10.20.32.114</td> 
   <td>2</td> 
   <td>2</td> 
  </tr> 
  <tr> 
   <td>zk3</td> 
   <td>10.20.32.126</td> 
   <td>3</td> 
   <td>3</td> 
  </tr> 
 </tbody> 
</table>

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

### ZooKeeper 安装部分 ###

1.  分别在三台机器上，写入host名称

修改 \\etc\\hosts

10.20.32.113 zk1
    10.20.32.114 zk2
    10.20.32.126 zk3

1.  [下载][Link 1]解压ZooKeeper，我用的是zookeeper-3.4.11.统一放到**/usr/local/zookeeper-3.4.11/**
2.  创建数据文件夹 ,写入zookeeper的ID到myid文件里，**注意每台机器都不一样,可参照上面的表格**，查看服务器 IP MYID的对应关系

mkdir /usr/local/zookeeper-3.4.11/data
    touch /usr/local/zookeeper-3.4.11/data/myid
    echo "1" > /usr/local/zookeeper-3.4.11/data/myid

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

1.  在zk的conf文件夹里，复制zoo\_sample.cfg为zoo.cfg，并修改端口（**可选，我这里使用的不是默认端口2181，而是1119**）

# The number of milliseconds of each tick
    tickTime=2000
    # The number of ticks that the initial 
    # synchronization phase can take
    initLimit=10
    # The number of ticks that can pass between 
    # sending a request and getting an acknowledgement
    syncLimit=5
    # the directory where the snapshot is stored.
    # do not use /tmp for storage, /tmp here is just 
    # example sakes.
    dataDir=/usr/local/zookeeper-3.4.11/data/
    # the port at which the clients will connect
    clientPort=1119
    # the maximum number of client connections.
    # increase this if you need to handle more clients
    #maxClientCnxns=60
    #
    # Be sure to read the maintenance section of the 
    # administrator guide before turning on autopurge.
    #
    # http://zookeeper.apache.org/doc/current/zookeeperAdmin.html#sc_maintenance
    #
    # The number of snapshots to retain in dataDir
    #autopurge.snapRetainCount=3
    # Purge task interval in hours
    # Set to "0" to disable auto purge feature
    #autopurge.purgeInterval=1
    
    server.1=zk1:5888:6888
    server.2=zk2:5888:6888
    server.3=zk3:5888:6888

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

修改 /etc/rc.local 配置ZooKeeper开机启动(**可选**),注意这里的JAVA\_HOME要指向你本机实际的JAVA目录，这里我使用yum安装的openjdk

yum  install  java-1.8.0-openjdk   java-1.8.0-openjdk-devel

/etc/rc.local

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.151-1.b12.el6_9.x86_64/jre
    /usr/local/zookeeper-3.4.11/bin/zkServer.sh start

下面就可以在三台机器上分别启动zk了

/usr/local/zookeeper-3.4.11/bin/zkServer.sh start

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

### Kafka安装部分 ###

1.  [下载][Link 2]解压Kafka 0.11.0.2，这里我统一安装到 **/usr/local/kafka\_2.11-0.11.0.2/**
2.  修改配置文件如下（以配置第一台kafka机器为例，配置很简单只需要修改配置文件的3处即可）

vi /usr/local/kafka_2.11-0.11.0.2/config/server.properties
    
    #修改，这里ID为1
    
    broker.id=1
    
    #这里指定kafak运行端口，可以填写IP或者机器名
    
    listeners=PLAINTEXT://zk1:9092
    
    #指定zk集群的连接地址
    zookeeper.connect=zk1:1119,zk2:1119,zk3:1119

1.  依次完成3个机器的配置修改。下面启动Kafka集群

/usr/local/kafka_2.11-0.11.0.2/bin/kafka-server-start.sh -daemon /usr/local/kafka_2.11-0.11.0.2/config/server.properties

另外可以把这个命令加入/etc/rc.local开机启动（可选）

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.151-1.b12.el6_9.x86_64/jre
    /usr/local/zookeeper-3.4.11/bin/zkServer.sh start
    /usr/local/kafka_2.11-0.11.0.2/bin/kafka-server-start.sh -daemon /usr/local/kafka_2.11-0.11.0.2/config/server.properties

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

### 验证测试Kafka ###

以下所以操作都是在**/usr/local/kafka\_2.11-0.11.0.2/bin**

cd /usr/local/kafka_2.11-0.11.0.2/bin

##### 首先创建topic，此处我的tpoic叫Mark\_test #####

/kafka-topics.sh  --create --zookeeper zk1:1119  --replication-factor 1 --partitions 1 --topic Mark_test

##### 在本机或者其他2台机器上查看一下是否能看到这个topic #####

./kafka-topics.sh --list --zookeeper zk1:1119 Mark_test

##### 在某一台机器上创建消息，运行命令以后可发送任意消息，此处为生产者 #####

./kafka-console-producer.sh --broker-list zk1:9092 --topic Mark_test

##### 在某一台机器上消费消息，运行命令以后可接受指定topic的消息，此处为消费者，可以在其他2台机器上运行此命令 #####

./kafka-console-consumer.sh --zookeeper localhost:1119 --topic Mark_test --from-beginning

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

### java客户端验证 ###

建立maven项目，pom.xml内容如下

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <groupId>my.kafka.test</groupId>
        <artifactId>demo1</artifactId>
        <version>1.0-SNAPSHOT</version>
        <dependencies>
            <dependency>
                <groupId>org.apache.kafka</groupId>
                <artifactId>kafka_2.11</artifactId>
                <version>0.11.0.2</version>
            </dependency>
        </dependencies>
    </project>

#### java生产者 ####

import java.util.Properties;
    
    import kafka.javaapi.producer.Producer;
    
    import kafka.producer.KeyedMessage;
    
    import kafka.producer.ProducerConfig;
    
    /** * Hello world! */
    
    public class KafkaProducer { 
    
        public final static String TOPIC = "Mark_test";
        private final Producer<String, String> producer;
    
        private KafkaProducer() {
    
            Properties props = new Properties();
    
            // 此处配置的是kafka的端口
    
            props.put("metadata.broker.list", "10.20.32.126:9092");
    
            // 配置value的序列化类
    
            props.put("serializer.class", "kafka.serializer.StringEncoder");
    
            // 配置key的序列化类
    
            props.put("key.serializer.class", "kafka.serializer.StringEncoder");
    
            // request.required.acks
    
            // 0, which means that the producer never waits for an acknowledgement
            // from the broker (the same behavior as 0.7). This option provides the
            // lowest latency but the weakest durability guarantees (some data will
            // be lost when a server fails).
    
            // 1, which means that the producer gets an acknowledgement after the
            // leader replica has received the data. This option provides better
            // durability as the client waits until the server acknowledges the
            // request as successful (only messages that were written to the
            // now-dead leader but not yet replicated will be lost).
    
            // -1, which means that the producer gets an acknowledgement after all
            // in-sync replicas have received the data. This option provides the
            // best durability, we guarantee that no messages will be lost as long
            // as at least one in sync replica remains.
    
            props.put("request.required.acks", "-1");
    
            producer = new Producer<String, String>(new ProducerConfig(props));
    
        }
    
        public static void main(String[] args)
    
        {
    
            new KafkaProducer().produce();
    
        }
    
        void produce() {
    
            int messageNo = 1000;
    
            final int COUNT = 10000;
    
            while (messageNo < COUNT) {
    
                String key = String.valueOf(messageNo);
    
                String data = "Current time is:" + System.currentTimeMillis() + " msg key: " + key;
    
                producer.send(new KeyedMessage<String, String>(TOPIC, key, data));
    
                System.out.println(data);
    
                messageNo++;
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
    
            }
    
        }
    
    }

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

#### java消费者 ####

import kafka.consumer.ConsumerConfig;
    import kafka.consumer.ConsumerIterator;
    import kafka.consumer.KafkaStream;
    import kafka.javaapi.consumer.ConsumerConnector;
    import kafka.serializer.StringDecoder;
    import kafka.utils.VerifiableProperties;
    
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    import java.util.Properties;
    
    public class KafkaConsumer { 
    
        private final ConsumerConnector consumer;
    
        public KafkaConsumer() {
    
            Properties props = new Properties();
    
            // zookeeper 配置
    
            props.put("metadata.broker.list", "10.20.32.126:9092");
    
            // group 代表一个消费组
    
            props.put("group.id", "jd-group");
    
            // zk连接超时
            props.put("zookeeper.connect", "zk1:1119,zk2:1119,zk3:1119");
            props.put("zookeeper.session.timeout.ms", "4000");
    
            props.put("zookeeper.sync.time.ms", "200");
    
            props.put("auto.commit.interval.ms", "1000");
    
            props.put("auto.offset.reset", "largest");
    
            // 序列化类
    
            props.put("serializer.class", "kafka.serializer.StringEncoder");
    
            ConsumerConfig config = new ConsumerConfig(props);
    
            consumer = kafka.consumer.Consumer.createJavaConsumerConnector(config);
    
        }
    
        public static void main(String[] args) {
    
            new KafkaConsumer().consume();
    
        }
    
        public void consume() {
    
            Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
    
            topicCountMap.put("Mark_test", new Integer(1));
    
            StringDecoder keyDecoder = new StringDecoder(new VerifiableProperties());
    
            StringDecoder valueDecoder = new StringDecoder(
                    new VerifiableProperties());
    
            Map<String, List<KafkaStream<String, String>>> consumerMap =
    
                    consumer.createMessageStreams(topicCountMap, keyDecoder, valueDecoder);
    
            KafkaStream<String, String> stream = consumerMap.get(
                    "Mark_test").get(0);
    
            ConsumerIterator<String, String> it = stream.iterator();
    
            while (it.hasNext())
    
                System.out.println(it.next().message());
    
        }
    
    }

#### 最后放上配合0.11版本的kafka的[java API][] ####

[Link 1]: https://mirrors.tuna.tsinghua.edu.cn/apache/zookeeper/
[Link 2]: http://mirrors.tuna.tsinghua.edu.cn/apache/kafka/0.11.0.2/kafka_2.11-0.11.0.2.tgz
[java API]: http://kafka.apache.org/0110/javadoc/index.html?org/apache/kafka/clients/producer/KafkaProducer.html