目录
1、安装kafka
1.1确认jdk是否安装OK
1.2下载&&安装kafka
1.3验证kafka
2、连接kafka
3、在java中操作kafka
1、安装kafka
1.1确认jdk是否安装Ok
java -version
1.2下载&&安装kafka
wget http://archive.apache.org/dist/zookeeper/zookeeper-3.4.14/zookeeper-3.4.14.tar.gz
mkdir zk-3.4.14
tar -xvzf zookeeper-3.4.14.tar.gz -C /home/lighthouse/zk-3.4.14
配置
进入该目录下的conf文件夹中。
zoo_sample.cfg是一个配置文件的样本，需要将这个文件复制并重命名为zoo.cfg:
cp zoo_sample.cfg zoo.cfg
修改配置文件：
vi zoo.cfg
配置环境变量, 使用vim打开etc目录下的profile文件：vim /etc/profile
在末尾配置环境变量，这里需要写入的是：
export ZOOKEEPER_HOME=/home/lighthouse/zk-3.4.14/zookeeper-3.4.14
export PATH=$PATH:$ZOOKEEPER_HOME/bin
写入信息并保存后，需要使配置文件生效，所用的命令为：source /etc/profile
启动zookeeper, 由于配置了环境变量，可以在系统中的任意目录执行启动zookeeper的命令，其执行的实际上是zookeeper的bin文件夹中的zkServer.sh的命令：zkServer.sh start
Zookeeper启动成功：
下载kafka2.2.1：
wget https://archive.apache.org/dist/kafka/2.2.1/kafka_2.12-2.2.1.tgz
解压：
tar -zxvf kafka_2.12-2.2.1.tgz
启动：
nohup bin/kafka-server-start.sh config/server.properties > output.txt &
其中server.properties文件内容如下：
1.3验证kafka
执行命令：bin/kafka-topics.sh –version
看不到版本号
2、连接kafka，并执行命令
2.1创建topic：执行命令：
bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test
2.2查看topic：执行命令：
bin/kafka-topics.sh --list --zookeeper localhost:2181
bin/kafka-topics.sh --list --zookeeper 43.138.0.199:2181
2.3使用kafka-console-producer.sh 发送消息：
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test
bin/kafka-console-producer.sh --broker-list 43.138.0.199:9092 --topic test
2.4使用kafka-console-consumer.sh消费消息：
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning
bin/kafka-console-consumer.sh --bootstrap-server 43.138.0.199:9092 --topic test --from-beginning
3、在java中操作kafka
pom.xml增加如下依赖：
<dependencies>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.11</artifactId>
<version>1.0.1</version>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>1.0.1</version>
</dependency>
</dependencies>Producer.java代码如下：
package com.hmblogs.backend.util;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.log4j.Logger;
import java.util.Properties;
/******************************************************
****** @ClassName : Producer.java
****** @author : milo ^ ^
****** @date : 2018 03 14 11:34
****** @version : v1.0.x
*******************************************************/
public class Producer {
static Logger log = Logger.getLogger(Producer.class);
private static final String TOPIC = "test";
private static final String BROKER_LIST = "43.138.0.199:9092";
private static KafkaProducer<String,String> producer = null;
/*
初始化生产者
*/
static {
Properties configs = initConfig();
producer = new KafkaProducer<String, String>(configs);
}
/*
初始化配置
*/
private static Properties initConfig(){
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BROKER_LIST);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
return properties;
}
public static void main(String[] args) throws InterruptedException {
//消息实体
ProducerRecord<String , String> record = null;
for (int i = 0; i < 3; i++) {
record = new ProducerRecord<String, String>(TOPIC, "value"+(int)(10*(Math.random())));
//发送消息
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (null != e){
System.out.println("send error" + e.getMessage());
}else {
System.out.println(String.format("offset:%s,partition:%s",recordMetadata.offset(),recordMetadata.partition()));
}
}
});
}
producer.close();
}
}
执行报错，如下：
send errorExpiring 3 record(s) for test-0: 30034 ms has passed since batch creation plus linger time
send errorExpiring 3 record(s) for test-0: 30034 ms has passed since batch creation plus linger time
send errorExpiring 3 record(s) for test-0: 30034 ms has passed since batch creation plus linger time
搜索资料，尝试解决
既然本地调不通，那我就发到linux机器里面去调试
改成ProducerController
代码如下：
package com.hmblogs.backend.controller;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Properties;
@RestController
@Slf4j
public class ProducerController {
private static final String TOPIC = "test";
private static final String BROKER_LIST = "43.138.0.199:9092";
private static KafkaProducer<String,String> producer = null;
/**
* sendMessage
* @return
*/
@GetMapping(value = "/sendMessage")
public void redisTestLock(){
log.info("sendMessage");
Properties configs = initConfig();
producer = new KafkaProducer<String, String>(configs);
send();
}
/*
初始化配置
*/
private Properties initConfig(){
Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,BROKER_LIST);
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,StringSerializer.class.getName());
return properties;
}
private void send() {
//消息实体
ProducerRecord<String , String> record = null;
for (int i = 0; i < 3; i++) {
record = new ProducerRecord<String, String>(TOPIC, "value："+(int)(10*(Math.random())));
//发送消息
producer.send(record, new Callback() {
@Override
public void onCompletion(RecordMetadata recordMetadata, Exception e) {
if (null != e){
log.error("send error:" + e.getMessage());
}else {
log.info(String.format("offset:%s,partition:%s",recordMetadata.offset(),recordMetadata.partition()));
}
}
});
}
producer.close();
}
}
执行clean install命令，打包成jar文件，上传到云主机里，然后启动
java -jar hmblogs.jar
访问GET接口
http://43.138.0.199:8081/sendMessage
查看hmblogs服务的日志
然后，重新查看该topic的消息
消费的验证也改成通过调用接口来实现
ConsumerController代码如下：
package com.hmblogs.backend.controller;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.Collections;
import java.util.Properties;
@RestController
@Slf4j
public class ConsumerController {
private static final String TOPIC = "test";
private static final String BROKER_LIST = "43.138.0.199:9092";
private static KafkaConsumer<String,String> consumer = null;
/**
* consumeMessage
* @return
*/
@GetMapping(value = "/consumeMessage")
public void consumeKafkaMessage(){
log.info("consumeMessage");
Properties configs = initConfig();
consumer = new KafkaConsumer<String, String>(configs);
consumer.subscribe(Collections.singletonList(TOPIC));
send();
}
private static Properties initConfig(){
Properties properties = new Properties();
properties.put("bootstrap.servers",BROKER_LIST);
properties.put("group.id","0");
properties.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("enable.auto.commit", "true");
properties.setProperty("auto.offset.reset", "earliest");
return properties;
}
private void send() {
while (true) {
ConsumerRecords<String, String> records = consumer.poll(10);
for (ConsumerRecord<String, String> record : records) {
log.info("Received message: key={}, value={}, partition={}, offset={}\n",
record.key(), record.value(), record.partition(), record.offset());
}
}
}
}
然后clean install，发到云主机里，调用如下GET接口：
http://43.138.0.199:8081/consumeMessage
hmblogs.log日志显示内容如下截图：文章来源：https://www.toymoban.com/news/detail-798838.html
文章来源地址https://www.toymoban.com/news/detail-798838.html
到了这里，关于spring boot学习第八篇:kafka的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！
