实现之前一定要把JDK,Zookeeper和Kafka都配置好
需要先配置下Kafka
在kafka的config目录下找到server.properties配置文件
image.png
把listeners和 advertised.listeners两处配置的注释去掉，可以根据需要配置连接的服务器外网IP和端口号，我这里演示选择的是本地localhost和默认端口9092
Kafka与SpringBoot进行整合
1.引入依赖
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
2.编写生产者和消费者
@RestController
public class KafkaController {
private static Logger logger = LoggerFactory.getLogger(KafkaController.class);
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
@GetMapping("/kafka")
public String testKafka() {
int iMax = 100;
for (int i = 1; i < iMax; i++) {
kafkaTemplate.send("test","key" + i, "data" + i);
}
return "success";
}
@KafkaListener(topics = "test")
public void receive(ConsumerRecord<?, ?> consumer) {
logger.info("{} - {}:{}", consumer.topic(), consumer.key(), consumer.value());
}
}
相关代码说明
KafkaTemplate这个类包装了个生产者Producer，来提供方便的发送数据到kafka的主题topic里面。
send()方法的源码，KafkaTemplate类中还重载了很多send()方法，有需要可以看看源码
public ListenableFuture<SendResult<K, V>> send(String topic, K key, @Nullable V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord(topic, key, data);
return this.doSend(producerRecord);
}
通过KafkaTemplate模板类发送数据。
kafkaTemplate.send(String topic, K key, V data),第一个入参是主题，第二个入参是发送的对象，第三个入参是发送的数据。通过@KafkaListener注解配置用户监听topics
配置文件application.yml
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
group-id: kafka2
auto-offset-reset: earliest
key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
producer:
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
bootstrap-servers：kafka服务器地址(可以多个)
consumer.group-id:指定一个默认的组名
不指定的话会报
java.lang.IllegalStateException: No group.id found in consumer config,
container properties, or @KafkaListener annotation;
a group.id is required when group management is used.
auto-offset-reset：自动偏移量
1.earliest ：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费
2.latest ：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据
3.none ：topic各分区都存在已提交的offset时，从offset后开始消费；只要有一个分区不存在已提交的offset，则抛出异常
这个属性也是必须配置的，不然也是会报错的
org.apache.kafka.common.config.ConfigException:
Invalid value for configuration auto.offset.reset:
String must be one of: latest, earliest, none
消息序列化和反序列化
在使用Kafka发送接收消息时，生产者producer端需要序列化，消费者consumer端需要反序列化，由于网络传输过来的是byte[]，只有反序列化后才能得到生产者发送的真实的消息内容。这样消息才能进行网络传输
consumer.key-deserializer和consumer.value-deserializer是消费者key/value反序列化
producer.key-deserializer和producer.value-deserializer是生产者key/value序列化
StringDeserializer是内置的字符串反序列化方式
public class StringDeserializer implements Deserializer<String> {
public String deserialize(String topic, byte[] data) {
try {
//如果数据为空，那么直接返回null即可,否则将byte[]反序列化，即转为String即可
return data == null ? null : new String(data, this.encoding);
} catch (UnsupportedEncodingException var4) {
throw new SerializationException("Error when deserializing byte[] to string due to unsupported encoding " + this.encoding);
}
}
......
}
StringSerializer是内置的字符串序列化方式
public class StringSerializer implements Serializer<String> {
public byte[] serialize(String topic, String data) {
try {
//如果数据为空，那么直接返回null即可,否则将String序列化，即转为byte[]即可
return data == null ? null : data.getBytes(this.encoding);
} catch (UnsupportedEncodingException var4) {
throw new SerializationException("Error when serializing string to byte[] due to unsupported encoding " + this.encoding);
}
}
......
}
在org.apache.kafka.common.serialization源码包中还提供了多种类型的序列化和反序列化方式
要自定义序列化方式，需要实现接口Serializer
要自定义反序列化方式，需要实现接口Deserializer
详细可以参考
https://blog.csdn.net/shirukai/article/details/82152172
启动项目进行测试
这是Kafka的消费者Consumer的配置信息，每个消费者都会输出该配置信息
auto.commit.interval.ms = 5000
auto.offset.reset = earliest
bootstrap.servers = [localhost:9092]
check.crcs = true
client.id =
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 = kafka2
heartbeat.interval.ms = 3000
interceptor.classes = []
internal.leave.group.on.close = true
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
send.buffer.bytes = 131072
session.timeout.ms = 10000
ssl.cipher.suites = null
ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
ssl.endpoint.identification.algorithm = https
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
2018-11-22 14:16:53.465 INFO 11980 --- [ main] o.a.kafka.common.utils.AppInfoParser : Kafka version : 2.0.0
2018-11-22 14:16:53.465 INFO 11980 --- [ main] o.a.kafka.common.utils.AppInfoParser : Kafka commitId : 3402a8361b734732
2018-11-22 14:16:57.664 INFO 11980 --- [ main] org.apache.kafka.clients.Metadata : Cluster ID: d3n7Snc2TFmSFcNsHjqgVw
访问http://localhost:8080/kafka,就可以看到控制台打印消息了
image.png
网友评论