kafka安装和使用java连接远程服务器进行消息的生成与消费

首先要使用kafka，要有jdk和zookeeper的环境

本文在阿里云的centos7环境上进行

jdk版本选择的是1.8.0_181

zookeeper的版本是3.4.12

kafka的版本是2.12-1.1.1

关于kafka命令的介绍 本文不介绍了 只介绍怎么搭建一个kafka单点服务器 以及怎么使用代码 远程连接kafka服务器

下载地址

kafka下载地址 :http://kafka.apache.org/downloads

zookeeper下载地址：https://zookeeper.apache.org/

jdk下载地址：https://www.oracle.com/technetwork/java/javase/downloads/jdk8-downloads-2133151.html

操作步骤

1、首先 使用tar命令对jdk进行解压
    tar -zxvf tar -zxvf jdk-8u181-linux-x64.tar.gz
    目录下面会多出一个jdk1.8.0_181  进入里面去  使用pwd命令查看绝对路径  并且复制找个路径
    最后进行jdk环境变量的配置
    编辑 vim /etc/profile文件
    在文件后面加上：
    export JAVA_HOME=（刚才pwd命令看到的路径）
    export CLASSPATH=.:${JAVA_HOME}/jre/lib/rt.jar:${JAVA_HOME}/lib/dt.jar:${JAVA_HOME}/            lib/    tools.jar
    export PATH=$PATH:${JAVA_HOME}/bin

    最后使用source /etc/profile 刷新文件

    使用java -version 查看环境变量是否配置成功

2、成功之后进行zookeeper的安装

    使用 tar -zxvf zookeeper-3.4.12.tar.gz 接下下载好的zookeeper安装包

    将zookeeper下的/conf/zookeeper.example改名成zoo.cfg
    使用mv 和cp命令都可以  然后vim这个文件 加上下面两行
    dataLogDir=/tmp/zookeeper-log #日志路径
    quorumListenOnAllIPs=true #在阿里云的服务器上保证外网可以访问到  刚开始没设置这个折腾了好久
3、最后，安装kafka
    使用 tar -zxvf kafka_2.12-1.1.1.tgz 解压下载好的kafka
    cd 到解压后的文件里面去  编辑配置文件  vim config/server.properties
    加上下面几行
    listeners=PLAINTEXT://:9092
    advertised.host.name=阿里云服务器公网ip #
    advertised.port=9092

    将zookeeper.connect的值改为阿里云的公网ip
    

至此，所有的环境的安装已经完成，下面使用kafka的命令进行消息的生成和消费

    首先cd到zookeeper的bin目录下  使用 ./zkServer.sh start 启动zookeeper
    再cd到kafka的bin目录下 使用 ./kafka-server-start.sh ../config/server.properties 启动kafka

    新建一个会话或者打开一个新的终端
    这时候使用jps命令  可以看到 Kafka和QuorumPeerMain表示启动全部成功，下面创建一个主题
    cd到kafka的bin目录下面，执行
    ./kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --     partitions 1 --topic Hello-world

    输出Created topic "Hello-world".  表示topic创建成功
    使用./kafka-topics.sh --list --zookeeper localhost:2181 查看主题的列表
    输出里面会含有Hello-world

    下面进行消息的生产和消费
    先启动生产者 ./kafka-console-producer.sh --broker-list 阿里云公网ip:9092 --topic Hello-        world
    会出现一个 >  类似于交互界面 这时候就可以生产消息了

    启动消费者 ./kafka-console-consumer.sh --zookeeper 阿里云公网ip:2181 --topic Hello-       world --from-beginning 

    这时候当生产者生产消息的时候  消费者这边就可以看到了  

    

在服务器上面进行消息的生产和消费就完成了 下面介绍怎么使用java代码进行远程连接kafka服务器

这个地方真的踩了好多好多坑、有次晚上下班搞到了快两点 百度、谷歌、维基、Stack Overflow 能找解决问题的地方都找了浪费了好多不必要的时间

    首先、新建一个Maven工程（此处不再多描述），在pom文件中加入kafka的依赖
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka_2.10</artifactId>
        <version>0.8.2.0</version>
    </dependency>

    新建一个KafkaProducerDemo和KafkaConsumerDemo类（名字可以自定义）：
    话不多说  上代码

    KafkaProducerDemo类：

            public class KafkaProducerDemo {
                public static void main(String[] args) {
                    //创建properties文件
                    Properties properties = new Properties();
                    //设置kafka服务器地址
                   properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "阿里云公网ip:9092");
                   //设置key进行序列化
                   properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                   //设置value进行序列化
                   properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
                   //创建消息生产者
                   KafkaProducer<String,String> producer = new KafkaProducer<>(properties);
                   //创建消息实体 制定主题、key、value
                   ProducerRecord<String,String> record = new ProducerRecord<>("Hello-world","haha","from java client");
                   //发送消息
                   producer.send(record);
                   System.out.println("消息发送成功");
                   //关闭生产者
                   producer.close();

                }
            }

    KafkaConsumerDemo类：

     public class KafkaConsumerDemo {

        public static void main(String[] args) {
            //新建配置文件
            Properties properties = new Properties();
            //设置kafka服务器地址
            properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,"阿里云公网ip:9092");
            //设置key的反序列化
            properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
            //设置value的反序列化
            properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,"org.apache.kafka.common.serialization.StringDeserializer");
            //设置groupid
            properties.put(ConsumerConfig.GROUP_ID_CONFIG,"test");

            //创建消费者对象
            KafkaConsumer<String,String> consumer = new KafkaConsumer<>(properties);
            //订阅主题
            consumer.subscribe(Arrays.asList("Hello-world"));

            while (true) {
                //消费消息
                ConsumerRecords<String, String> records = consumer.poll(100);
                for (ConsumerRecord<String, String> record : records)

                    System.out.printf("offset = %d, key = %s, value = %s\n",
                            record.offset(), record.key(), record.value());
            }
        }
    }

    上面就是连接kafka远程服务器代码

    

但是上述过程做完之后还是不能正确运行、这个地方折腾了好久、最后在哪里看到解决的办法记不大清了

关闭了防火墙、没什么用，不知道什么鬼。 搞得我头皮发麻

原文地址：https://www.cnblogs.com/evildoerdb/p/9650808.html