Kafka集群（ 2.8.0版本）之四：Java简单实现生产者和消费者
Kafka集群（ 2.8.0版本）之四：Java简单实现生产者和消费者
程裕强
发布于 2021-09-09 15:52:31
发布于 2021-09-09 15:52:31
文章被收录于专栏：大数据学习笔记
1、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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.5.4</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.example</groupId>
<artifactId>kafkaDemo</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>kafkaDemo</name>
<description>kafkaDemo</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>2.8.0</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>2、application.properties
server.port=7780
server.servlet.context-path=/kafka3、生产者和消费者
package com.example.kafkademo;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RestController;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;
@RestController
@SpringBootApplication
public class KafkaDemoApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaDemoApplication.class, args);
}
@GetMapping("/send/{key}/{value}")
public String send(@PathVariable("key") String key, @PathVariable("value") String value) {
//配置信息
Properties props = new Properties();
//kafka服务器地址
props.put("bootstrap.servers", "10.17.12.158:9092");
//设置数据key和value的序列化处理类
props.put("key.serializer", StringSerializer.class);
props.put("value.serializer", StringSerializer.class);
//创建生产者实例
KafkaProducer<String,String> producer = new KafkaProducer<>(props);
ProducerRecord record = new ProducerRecord<String, String>("demo", key, value);
//发送记录
producer.send(record);
producer.close();
return "Ok";
}
@GetMapping("/receive/{topic}")
public void receive(@PathVariable("topic") String topic) {
System.out.println("topic="+topic);
//配置信息
Properties props = new Properties();
//kafka服务器地址
props.put("bootstrap.servers", "node1:9092,node2:9092,node3:9092");
//必须指定消费者组
props.put("group.id", "test");
//设置数据key和value的序列化处理类
props.put("key.deserializer", StringDeserializer.class);
props.put("value.deserializer", StringDeserializer.class);
//创建消息者实例
KafkaConsumer<String,String> consumer = new KafkaConsumer<>(props);
//订阅topic1的消息
consumer.subscribe(Arrays.asList(topic));
//到服务器中读取记录
while (true) {
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(2000));
System.out.println("获取消息条数："+records.count());
for (ConsumerRecord<String, String> record : records) {
System.out.println("key:" + record.key() + "" + ",value:" + record.value());
}
}
}
}4、编译Jar包
（1）在IDEA底部找到Terminal
（2）确认项目根目录，执行mvn clean package
（3）编译成功，看到BUILD SUCCESS
（4）找到编译好的jar包
在这里插入图片描述
在这里插入图片描述
5、执行Jar包
（1）上传Jar包到服务器 （2）执行jar包
[root@node1 ~]# ls jar
kafkaDemo-0.0.1-SNAPSHOT.jar
[root@node1 jar]# java -jar /root/jar/kafkaDemo-0.0.1-SNAPSHOT.jar
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.5.4)
2021-09-06 21:50:46.475 INFO 20561 --- [ main] c.e.kafkademo.KafkaDemoApplication : Starting KafkaDemoApplication v0.0.1-SNAPSHOT using Java 1.8.0_161 on node1 with PID 20561 (/root/jar/kafkaDemo-0.0.1-SNAPSHOT.jar started by root in /root/jar)
2021-09-06 21:50:46.493 INFO 20561 --- [ main] c.e.kafkademo.KafkaDemoApplication : No active profile set, falling back to default profiles: default
2021-09-06 21:50:48.467 INFO 20561 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat initialized with port(s): 7780 (http)
2021-09-06 21:50:48.494 INFO 20561 --- [ main] o.apache.catalina.core.StandardService : Starting service [Tomcat]
2021-09-06 21:50:48.495 INFO 20561 --- [ main] org.apache.catalina.core.StandardEngine : Starting Servlet engine: [Apache Tomcat/9.0.52]
2021-09-06 21:50:48.604 INFO 20561 --- [ main] o.a.c.c.C.[Tomcat].[localhost].[/kafka] : Initializing Spring embedded WebApplicationContext
2021-09-06 21:50:48.605 INFO 20561 --- [ main] w.s.c.ServletWebServerApplicationContext : Root WebApplicationContext: initialization completed in 1981 ms
2021-09-06 21:50:49.283 INFO 20561 --- [ main] o.s.b.w.embedded.tomcat.TomcatWebServer : Tomcat started on port(s): 7780 (http) with context path '/kafka'
2021-09-06 21:50:49.301 INFO 20561 --- [ main] c.e.kafkademo.KafkaDemoApplication : Started KafkaDemoApplication in 3.594 seconds (JVM running for 4.398)（3）调用消费者程序
[root@node3 ~]# curl 10.17.12.158:7780/kafka/receive/demo注意这里只是调用消费者程序，对应的输出需要在启动Jar包终端，也就是（2）
（4）调用生产者程序
[root@node1 ~]# curl 10.17.12.158:7780/kafka/send/key1/value1
Ok[root@node1 ~]# curl 10.17.12.158:7780/kafka/send/key2/value2
Ok[root@node1 ~]# curl 10.17.12.158:7780/kafka/send/key3/value3（5）查看启动Jar包终端
2021-09-06 21:51:00.083 INFO 20561 --- [nio-7780-exec-1] o.a.c.c.C.[Tomcat].[localhost].[/kafka] : Initializing Spring DispatcherServlet 'dispatcherServlet'
2021-09-06 21:51:00.083 INFO 20561 --- [nio-7780-exec-1] o.s.web.servlet.DispatcherServlet : Initializing Servlet 'dispatcherServlet'
2021-09-06 21:51:00.086 INFO 20561 --- [nio-7780-exec-1] o.s.web.servlet.DispatcherServlet : Completed initialization in 3 ms
2021-09-06 21:51:00.198 INFO 20561 --- [nio-7780-exec-1] o.a.k.clients.producer.ProducerConfig : ProducerConfig values:
acks = 1
batch.size = 16384
bootstrap.servers = [10.17.12.158:9092]
buffer.memory = 33554432
client.dns.lookup = use_all_dns_ips
client.id = producer-1
compression.type = none
connections.max.idle.ms = 540000
delivery.timeout.ms = 120000
enable.idempotence = false
interceptor.classes = []
internal.auto.downgrade.txn.commit = false
key.serializer = class org.apache.kafka.common.serialization.StringSerializer
linger.ms = 0
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.max.age.ms = 300000
metadata.max.idle.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retries = 2147483647
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
security.providers = null
send.buffer.bytes = 131072
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.2
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
transaction.timeout.ms = 60000
transactional.id = null
value.serializer = class org.apache.kafka.common.serialization.StringSerializer
2021-09-06 21:51:00.415 INFO 20561 --- [nio-7780-exec-1] o.a.kafka.common.utils.AppInfoParser : Kafka version: 2.8.0
2021-09-06 21:51:00.417 INFO 20561 --- [nio-7780-exec-1] o.a.kafka.common.utils.AppInfoParser : Kafka commitId: ebb1d6e21cc92130
2021-09-06 21:51:00.417 INFO 20561 --- [nio-7780-exec-1] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1630936260413
2021-09-06 21:51:01.128 INFO 20561 --- [ad | producer-1] org.apache.kafka.clients.Metadata : [Producer clientId=producer-1] Cluster ID: R99s2cu-SoWrPuhulozybg
2021-09-06 21:51:01.153 INFO 20561 --- [nio-7780-exec-1] o.a.k.clients.producer.KafkaProducer : [Producer clientId=producer-1] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
2021-09-06 21:51:01.214 INFO 20561 --- [nio-7780-exec-1] org.apache.kafka.common.metrics.Metrics : Metrics scheduler closed
2021-09-06 21:51:01.215 INFO 20561 --- [nio-7780-exec-1] org.apache.kafka.common.metrics.Metrics : Closing reporter org.apache.kafka.common.metrics.JmxReporter
2021-09-06 21:51:01.215 INFO 20561 --- [nio-7780-exec-1] org.apache.kafka.common.metrics.Metrics : Metrics reporters closed
2021-09-06 21:51:01.217 INFO 20561 --- [nio-7780-exec-1] o.a.kafka.common.utils.AppInfoParser : App info kafka.producer for producer-1 unregistered
topic=demo
2021-09-06 21:51:08.241 INFO 20561 --- [nio-7780-exec-2] o.a.k.clients.consumer.ConsumerConfig : ConsumerConfig values:
allow.auto.create.topics = true
auto.commit.interval.ms = 5000
auto.offset.reset = latest
bootstrap.servers = [node1:9092, node2:9092, node3:9092]
check.crcs = true
client.dns.lookup = use_all_dns_ips
client.id = consumer-test-1
client.rack =
connections.max.idle.ms = 540000
default.api.timeout.ms = 60000
enable.auto.commit = true
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = test
group.instance.id = null
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
internal.throw.on.fetch.stable.offset.unsupported = false
isolation.level = read_uncommitted
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 500
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
security.providers = null
send.buffer.bytes = 131072
session.timeout.ms = 10000
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.2
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
2021-09-06 21:51:08.322 INFO 20561 --- [nio-7780-exec-2] o.a.kafka.common.utils.AppInfoParser : Kafka version: 2.8.0
2021-09-06 21:51:08.322 INFO 20561 --- [nio-7780-exec-2] o.a.kafka.common.utils.AppInfoParser : Kafka commitId: ebb1d6e21cc92130
2021-09-06 21:51:08.322 INFO 20561 --- [nio-7780-exec-2] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1630936268321
2021-09-06 21:51:08.324 INFO 20561 --- [nio-7780-exec-2] o.a.k.clients.consumer.KafkaConsumer : [Consumer clientId=consumer-test-1, groupId=test] Subscribed to topic(s): demo
2021-09-06 21:51:08.336 INFO 20561 --- [nio-7780-exec-2] org.apache.kafka.clients.Metadata : [Consumer clientId=consumer-test-1, groupId=test] Cluster ID: R99s2cu-SoWrPuhulozybg
2021-09-06 21:51:08.337 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Discovered group coordinator node3:9092 (id: 2147483644 rack: null)
2021-09-06 21:51:08.342 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-test-1, groupId=test] (Re-)joining group
2021-09-06 21:51:08.363 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-test-1, groupId=test] (Re-)joining group
2021-09-06 21:51:08.367 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Successfully joined group with generation Generation{generationId=22, memberId='consumer-test-1-1d435ff2-81b6-488d-9514-2ccb3be07970', protocol='range'}
2021-09-06 21:51:08.370 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Finished assignment for group at generation 22: {consumer-test-1-1d435ff2-81b6-488d-9514-2ccb3be07970=Assignment(partitions=[demo-0, demo-1, demo-2])}
2021-09-06 21:51:08.379 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.AbstractCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Successfully synced group in generation Generation{generationId=22, memberId='consumer-test-1-1d435ff2-81b6-488d-9514-2ccb3be07970', protocol='range'}
2021-09-06 21:51:08.380 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Notifying assignor about the new Assignment(partitions=[demo-0, demo-1, demo-2])
2021-09-06 21:51:08.384 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Adding newly assigned partitions: demo-0, demo-1, demo-2
2021-09-06 21:51:08.402 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Setting offset for partition demo-1 to the committed offset FetchPosition{offset=10, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[node3:9092 (id: 3 rack: null)], epoch=0}}
2021-09-06 21:51:08.404 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Setting offset for partition demo-2 to the committed offset FetchPosition{offset=11, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[node1:9092 (id: 1 rack: null)], epoch=0}}
2021-09-06 21:51:08.405 INFO 20561 --- [nio-7780-exec-2] o.a.k.c.c.internals.ConsumerCoordinator : [Consumer clientId=consumer-test-1, groupId=test] Setting offset for partition demo-0 to the committed offset FetchPosition{offset=12, offsetEpoch=Optional[0], currentLeader=LeaderAndEpoch{leader=Optional[node2:9092 (id: 2 rack: null)], epoch=0}}
获取消息条数：1
key:key1,value:value1
获取消息条数：0
获取消息条数：0
2021-09-06 21:51:14.172 INFO 20561 --- [nio-7780-exec-3] o.a.k.clients.producer.ProducerConfig : ProducerConfig values:
acks = 1
batch.size = 16384
bootstrap.servers = [10.17.12.158:9092]
buffer.memory = 33554432
client.dns.lookup = use_all_dns_ips
client.id = producer-2
compression.type = none
connections.max.idle.ms = 540000
delivery.timeout.ms = 120000
enable.idempotence = false
interceptor.classes = []
internal.auto.downgrade.txn.commit = false
key.serializer = class org.apache.kafka.common.serialization.StringSerializer
linger.ms = 0
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.max.age.ms = 300000
metadata.max.idle.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retries = 2147483647
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
security.providers = null
send.buffer.bytes = 131072
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.2
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
transaction.timeout.ms = 60000
transactional.id = null
value.serializer = class org.apache.kafka.common.serialization.StringSerializer
2021-09-06 21:51:14.181 INFO 20561 --- [nio-7780-exec-3] o.a.kafka.common.utils.AppInfoParser : Kafka version: 2.8.0
2021-09-06 21:51:14.181 INFO 20561 --- [nio-7780-exec-3] o.a.kafka.common.utils.AppInfoParser : Kafka commitId: ebb1d6e21cc92130
2021-09-06 21:51:14.181 INFO 20561 --- [nio-7780-exec-3] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1630936274181
2021-09-06 21:51:14.190 INFO 20561 --- [ad | producer-2] org.apache.kafka.clients.Metadata : [Producer clientId=producer-2] Cluster ID: R99s2cu-SoWrPuhulozybg
2021-09-06 21:51:14.192 INFO 20561 --- [nio-7780-exec-3] o.a.k.clients.producer.KafkaProducer : [Producer clientId=producer-2] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
获取消息条数：1
key:key2,value:value2
2021-09-06 21:51:14.208 INFO 20561 --- [nio-7780-exec-3] org.apache.kafka.common.metrics.Metrics : Metrics scheduler closed
2021-09-06 21:51:14.209 INFO 20561 --- [nio-7780-exec-3] org.apache.kafka.common.metrics.Metrics : Closing reporter org.apache.kafka.common.metrics.JmxReporter
2021-09-06 21:51:14.209 INFO 20561 --- [nio-7780-exec-3] org.apache.kafka.common.metrics.Metrics : Metrics reporters closed
2021-09-06 21:51:14.209 INFO 20561 --- [nio-7780-exec-3] o.a.kafka.common.utils.AppInfoParser : App info kafka.producer for producer-2 unregistered
获取消息条数：0
2021-09-06 21:51:17.074 INFO 20561 --- [nio-7780-exec-4] o.a.k.clients.producer.ProducerConfig : ProducerConfig values:
acks = 1
batch.size = 16384
bootstrap.servers = [10.17.12.158:9092]
buffer.memory = 33554432
client.dns.lookup = use_all_dns_ips
client.id = producer-3
compression.type = none
connections.max.idle.ms = 540000
delivery.timeout.ms = 120000
enable.idempotence = false
interceptor.classes = []
internal.auto.downgrade.txn.commit = false
key.serializer = class org.apache.kafka.common.serialization.StringSerializer
linger.ms = 0
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.max.age.ms = 300000
metadata.max.idle.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.recording.level = INFO
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.max.ms = 1000
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retries = 2147483647
retry.backoff.ms = 100
sasl.client.callback.handler.class = null
sasl.jaas.config = null
sasl.kerberos.kinit.cmd = /usr/bin/kinit
sasl.kerberos.min.time.before.relogin = 60000
sasl.kerberos.service.name = null
sasl.kerberos.ticket.renew.jitter = 0.05
sasl.kerberos.ticket.renew.window.factor = 0.8
sasl.login.callback.handler.class = null
sasl.login.class = null
sasl.login.refresh.buffer.seconds = 300
sasl.login.refresh.min.period.seconds = 60
sasl.login.refresh.window.factor = 0.8
sasl.login.refresh.window.jitter = 0.05
sasl.mechanism = GSSAPI
security.protocol = PLAINTEXT
security.providers = null
send.buffer.bytes = 131072
socket.connection.setup.timeout.max.ms = 30000
socket.connection.setup.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2]
ssl.endpoint.identification.algorithm = https
ssl.engine.factory.class = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.certificate.chain = null
ssl.keystore.key = null
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLSv1.2
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.certificates = null
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
transaction.timeout.ms = 60000
transactional.id = null
value.serializer = class org.apache.kafka.common.serialization.StringSerializer
2021-09-06 21:51:17.082 INFO 20561 --- [nio-7780-exec-4] o.a.kafka.common.utils.AppInfoParser : Kafka version: 2.8.0
2021-09-06 21:51:17.083 INFO 20561 --- [nio-7780-exec-4] o.a.kafka.common.utils.AppInfoParser : Kafka commitId: ebb1d6e21cc92130
2021-09-06 21:51:17.083 INFO 20561 --- [nio-7780-exec-4] o.a.kafka.common.utils.AppInfoParser : Kafka startTimeMs: 1630936277082
2021-09-06 21:51:17.092 INFO 20561 --- [ad | producer-3] org.apache.kafka.clients.Metadata : [Producer clientId=producer-3] Cluster ID: R99s2cu-SoWrPuhulozybg
2021-09-06 21:51:17.093 INFO 20561 --- [nio-7780-exec-4] o.a.k.clients.producer.KafkaProducer : [Producer clientId=producer-3] Closing the Kafka producer with timeoutMillis = 9223372036854775807 ms.
获取消息条数：1
key:key3,value:value3
2021-09-06 21:51:17.104 INFO 20561 --- [nio-7780-exec-4] org.apache.kafka.common.metrics.Metrics : Metrics scheduler closed
2021-09-06 21:51:17.105 INFO 20561 --- [nio-7780-exec-4] org.apache.kafka.common.metrics.Metrics : Closing reporter org.apache.kafka.common.metrics.JmxReporter
2021-09-06 21:51:17.105 INFO 20561 --- [nio-7780-exec-4] org.apache.kafka.common.metrics.Metrics : Metrics reporters closed
2021-09-06 21:51:17.105 INFO 20561 --- [nio-7780-exec-4] o.a.kafka.common.utils.AppInfoParser : App info kafka.producer for producer-3 unregistered
获取消息条数：0
获取消息条数：0
获取消息条数：0本文参与 腾讯云自媒体同步曝光计划，分享自作者个人站点/博客。
原始发表：2021/09/06 ，如有侵权请联系 cloudcommunity@tencent.com 删除
评论
登录后参与评论
推荐阅读
目录
腾讯云开发者
