测试三个方面,发送消息,接收消息,以及接收消息时bean的注入.
测试项目GitHub地址:
git@github.com:wudonghua/Java-Kafka-SpringBoot-API-test.git把刚才的项目打包到本地仓库:
-Dmaven.home=E:\apache-maven-3.2.5 -Dclassworlds.conf=E:\apache-maven-3.2.5\bin\m2.conf "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 2017.2.6\lib\idea_rt.jar=54786:C:\Program Files\JetBrains\IntelliJ IDEA 2017.2.6\bin" -Dfile.encoding=UTF-8 -classpath E:\apache-maven-3.2.5\boot\plexus-classworlds-2.5.2.jar org.codehaus.classworlds.Launcher -Didea.version=2017.2.6 -s E:\apache-maven-3.2.5\conf\settings.xml -Dmaven.repo.local=E:\repo -DskipTests=true install在测试工程里面添加jar包:
<dependency>
<groupId>Riven.kafka</groupId>
<artifactId>Java-SpringBoot-Kafka-API</artifactId>
<version>1.1.0-SNAPSHOT</version>
</dependency>发送消息:
配置值内容:
Riven:
kafka:
producer:
bootstrapServers: 服务器列表 #必填
retries: 99
acks: 接受策略
batchSize: 99
lingerMs: 99
bufferMemory: 99
keySerializer: org.apache.kafka.common.serialization.StringSerializer
valueSerializer: org.apache.kafka.common.serialization.StringSerializer实现类
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import java.time.LocalDateTime;
/**
* @Author dw07-Riven770[wudonghua@gznb.com]
* @Date 2017/12/1417:37
*/
@Component
@EnableScheduling
public class Producer {
@Autowired
private KafkaTemplate kafkaTemplate;
@Scheduled(fixedDelay = 6000)
private void sendMsg(){
kafkaTemplate.send("topic-test", "hello"+ LocalDateTime.now().toString());
}
}接收消息及Bean注入
配置内容:
Riven:
kafka:
consumer:
bootstrapServers: 服务器列表 #必填
enableAutoCommit: true
autoCommitIntervalMs: 99
sessionTimeoutMs: 99
fetchMinBytes: 99
maxPollRecords: 99
groupId: java #必填
autoOffseReset: latest
keySerializer: org.apache.kafka.common.serialization.StringDeserializer
valueSerializer: org.apache.kafka.common.serialization.StringDeserializer
consumerAmount: 99
PollTimeout: 99
topics[0]: test_group #必填实现类:
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import riven.kafka.api.listener.IKafkaListener;
import java.util.Optional;
/**
* @Author dw07-Riven770[wudonghua@gznb.com]
* @Date 2017/12/1417:37
*/
@Component
public class Consumer implements IKafkaListener{
@Autowired
private TestBean testBean;
@Override
public void listener(ConsumerRecord<?, ?> record) {
Optional<?> value = Optional.ofNullable(record.value());
String s = (String) value.get();
System.out.println("消费者"+s);
testBean.test();
}
}注入bean:
import org.springframework.stereotype.Component;
/**
* @Author dw07-Riven770[wudonghua@gznb.com]
* @Date 2017/12/1417:37
*/
@Component
public class TestBean {
public void test(){
System.out.println("DoSomeThing....");
}
}测试报文:
producer:启动报文
2017-12-15 17:15:31.332 INFO 16904 --- [pool-1-thread-1] o.a.k.clients.producer.ProducerConfig : ProducerConfig values:
acks = -1
batch.size = 4096
block.on.buffer.full = false
bootstrap.servers = [xx.xx.30.205:9092, xx.xx.30.206:9092]
buffer.memory = 40960
client.id = producer-1
compression.type = none
connections.max.idle.ms = 540000
interceptor.classes = null
key.serializer = class org.apache.kafka.common.serialization.StringSerializer
linger.ms = 1
max.block.ms = 60000
max.in.flight.requests.per.connection = 5
max.request.size = 1048576
metadata.fetch.timeout.ms = 60000
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.sample.window.ms = 30000
partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
receive.buffer.bytes = 32768
reconnect.backoff.ms = 50
request.timeout.ms = 30000
retries = 0
retry.backoff.ms = 100
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.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
timeout.ms = 30000
value.serializer = class org.apache.kafka.common.serialization.StringSerializer
producer发送报文:
2017-12-15 17:15:32.794 INFO 16904 --- [ad | producer-1] r.k.api.producer.SimpleProducerListener : 消息发送成功!
with key=【key】 and value=【Hello2017-12-15T17:15:31.328】 to topic 【topic-test】 send result: topicPartition【topic-test-1】 offset 【28】
consumer启动报文:
2017-12-15 17:15:31.299 INFO 16904 --- [ main] o.a.k.clients.consumer.ConsumerConfig : ConsumerConfig values:
auto.commit.interval.ms = 100
auto.offset.reset = latest
bootstrap.servers = [xx.xx.30.205:9092, xx.xx.30.206:9092]
check.crcs = true
client.id = consumer-3
connections.max.idle.ms = 540000
enable.auto.commit = false
exclude.internal.topics = true
fetch.max.bytes = 52428800
fetch.max.wait.ms = 500
fetch.min.bytes = 1
group.id = group_test
heartbeat.interval.ms = 3000
interceptor.classes = null
key.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
max.partition.fetch.bytes = 1048576
max.poll.interval.ms = 300000
max.poll.records = 300
metadata.max.age.ms = 300000
metric.reporters = []
metrics.num.samples = 2
metrics.sample.window.ms = 30000
partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]
receive.buffer.bytes = 65536
reconnect.backoff.ms = 50
request.timeout.ms = 305000
retry.backoff.ms = 100
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.mechanism = GSSAPI
security.protocol = PLAINTEXT
send.buffer.bytes = 131072
session.timeout.ms = 15000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = null
ssl.key.password = null
ssl.keymanager.algorithm = SunX509
ssl.keystore.location = null
ssl.keystore.password = null
ssl.keystore.type = JKS
ssl.protocol = TLS
ssl.provider = null
ssl.secure.random.implementation = null
ssl.trustmanager.algorithm = PKIX
ssl.truststore.location = null
ssl.truststore.password = null
ssl.truststore.type = JKS
value.deserializer = class org.apache.kafka.common.serialization.StringDeserializer
consumer消费报文:
消费者Hello2017-12-15T17:16:13.581
DoSomeThing....
