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.println("消息的主题是：" + record.topic()+"，消息的key是：" + record.key()+",消息的value是："+record.value());
}
}
}
上面就是连接kafka远程服务器代码
但是上述过程做完之后还是不能正确运行、这个地方折腾了好久、最后在哪里看到解决的办法记不大清了
就是要阿里云服务器服务安全设置里面加个规则 将2181和9092端口开放就可以，但是我中间也使用命令的方式关闭了防火墙、没什么用，不知道什么鬼。 搞得我头皮发麻
