在Centos6或者7上安装Kafka最新版

1、官网

　　http://kafka.apache.org/downloads.html

2、Kafka简介

　　Kafka是由Apache软件基金会开发的一个开源流处理平台，由Scala和Java编写。Kafka是一种高吞吐量的分布式发布订阅消息系统，它能够处理消费者规模的网站中的全部动做流数据。 这种动做（网页浏览，搜索和其余用户的行动）是在现代网络上的许多社会功能的一个关键因素。 这些数据一般是因为吞吐量的要求而经过处理日志和日志聚合来解决。 对于像Hadoop的同样的日志数据和离线分析系统，但又要求实时处理的限制，这是一个可行的解决方案。Kafka的目的是经过Hadoop的并行加载机制来统一线上和离线的消息处理，也是为了经过集群来提供实时的消息。

　　相似的组件还有：Azure的ServiceBus、RabbitMQ等，据网上描述，Kafka比RabbitMQ性能强。

3、安装

　　一、安装Java

yum install java-1.8.0-openjdk.x86_64

　　二、配置JAVA环境变量

export JAVA_HOME=/usr/lib/jvm/java-1.8.0-openjdk-1.8.0.171-8.b10.el7_5.x86_64/jre
export JRE_HOME=${JAVA_HOME}/jre
export CLASSPATH=.:${JAVA_HOME}/lib:${JRE_HOME}/lib
export PATH=${JAVA_HOME}/bin:$PATH

　　说明：配置时注意JAVA_HOME后面要加到/jre，这个比较特殊。另外，红色区域能够换成您对应的安装版本的路径。

　　三、下载Kafka：http://kafka.apache.org/downloads.html

cd /opt
wget http://mirrors.hust.edu.cn/apache/kafka/1.1.0/kafka_2.12-1.1.0.tgz

　　四、解压并进入目录

tar -zvxf ./kafka_2.12-1.1.0.tgz
cd kafka_2.12-1.1.0

　　五、启动Zookeeper

　　使用安装包中的脚本启动单节点Zookeeper 实例：（参数-daemon表示后台运行）

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

　　六、启动Kafka服务

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

　　七、建立一个测试的Topic

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

　　查看Topic：

bin/kafka-topics.sh --list --zookeeper localhost:2181

 

　　八、产生消息

bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
>hello xingzhu
>hello sindrol

 

　　九、消费消息

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

　　好了，到此，单台Kafka已经安装完成。

4、集群配置

　　一、单机多broker 集群配置

　　　  利用单节点部署多个broker。 不一样的broker 设置不一样的 id，监听端口及日志目录。 例如：

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

　　　　编辑配置：

config/server-1.properties:
    broker.id=1
    port=9093
    log.dir=/tmp/kafka-logs-1

　　　　启动Kafka服务：

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

　　　　启动多个服务后，能够参考第三节内容，产生和消费消息。

 

　　二、多机多broker 集群配置

　　　　分别在多个节点按上述方式安装Kafka，配置启动多个Zookeeper 实例。 例如： 在10.4.253.22，10.4.253.23，10.4.253.24三台机器部署，Zookeeper配置以下：

initLimit=5
syncLimit=2
server.1=10.4.253.22:2888:3888
server.2=10.4.253.23:2888:3888
server.3=10.4.253.24:2888:3888

　　　　分别配置多个机器上的Kafka服务 设置不一样的broke id，zookeeper.connect设置以下:

zookeeper.connect=10.4.253.22:2181,10.4.253.23:2181,10.4.253.24:2181

　　　　启动Zookeeper与Kafka服务，按上文方式产生和消费消息，验证集群功能。

 

 

5、外网访问

　　安装完成并启动后，若是想要外网经过外网IP访问，须要在config/service.properites中添加以下修改：

advertised.listeners=PLAINTEXT://xxx.xxx.xxx.xxx:port

　　修改完成后，从新启动Kafka服务。

 

6、C#调用

　　引入库：kafka-net(https://github.com/Jroland/kafka-net)

　　一、模拟消费端

using KafkaNet;
using KafkaNet.Model;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace KafkaClientDemo.Consumer
{
    class Program
    {
        static void Main(string[] args)
        {
            var options = new KafkaOptions(new Uri("http://42.159.154.132:9092"));
            var router = new BrokerRouter(options);
            
            var consumer = new KafkaNet.Consumer(new ConsumerOptions("test", router));

            Console.WriteLine("waiting ...");
            //Consume returns a blocking IEnumerable (ie: never ending stream)
            foreach (var message in consumer.Consume())
            {
                Console.WriteLine("Response: P{0},O{1} : {2}", message.Meta.PartitionId, message.Meta.Offset, Encoding.UTF8.GetString(message.Value, 0, message.Value.Length));
            }
        }
    }
}

　　二、模拟消息生产端

using KafkaNet;
using KafkaNet.Model;
using KafkaNet.Protocol;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace KafkaClientDemo
{
    class Program
    {
        static void Main(string[] args)
        {
            var options = new KafkaOptions(new Uri("http://42.159.154.132:9092"));
            var router = new BrokerRouter(options);
            var client = new Producer(router);
            //var topic = client.GetTopic("test");

            using (client)
            {
                while (true)
                {
                    Console.Write(">");
                    var text = Console.ReadLine();
                    client.SendMessageAsync("testTopic", new[] { new Message(text, "key_" + DateTime.Now.Ticks) }).Wait();
                  
                    if (text == "exit")
                        break;
                }
            }
        }
    }
}

 

　　运行效果：

 

　　

　　我在部署和学习时，参考了https://www.mtyun.com/library/how-to-install-kafka-on-centos7文章，很是感谢原做者分享。