重要的地方说下，算是给自己提醒，也给阅读者凑合着看看吧；
（1）序列化、反序列化；
注意看这个文章
https://www.jianshu.com/p/5da86afed228
很多网上的例子都是 推送字符串对象的，大多也都算是DEMO，要转实战成对象进行传递，需要序列化和反序列化，否则肯定要报错
（2）端口号问题
按照我第一部分的配置，配置了 zookeeper 后，正常启动后8080会被zookeeper占用，从而导致整个SpringBoot项目启动失败，报端口被占用异常；
（3）KAFKA作为分布式组件，肯定有博大精深的地方，应用场景很多，需要学习学习，仔细研究
可参考 https://www.jianshu.com/p/a64defb44a23
我的参考代码，仅供参考
package com.kafkastudy.kafka01;
import com.kafkastudy.kafka01.producer.UserProducer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
import javax.annotation.PostConstruct;
@SuppressWarnings("SpringJavaInjectionPointsAutowiringInspection")
@SpringBootApplication
@ComponentScan(basePackages = {"com.kafkastudy.kafka01.*"})
public class Kafka01Application {
@Autowired
private UserProducer kafkaSender;
@PostConstruct
public void init(){
System.out.println("----------start ----------------");
for (int i = 0; i < 10; i++) {
//调用消息发送类中的消息发送方法
kafkaSender.sendData(i);
}
System.out.println("----------end ----------------");
}
public static void main(String[] args) {
SpringApplication.run(Kafka01Application.class, args);
}
}
属性文件配置如下，调整了条数和数据容量
server.port=8081 #============== kafka =================== # 指定kafka server的地址，集群配多个，中间，逗号隔开 spring.kafka.bootstrap-servers=127.0.0.1:9092 #=============== provider ======================= # 写入失败时，重试次数。当leader节点失效，一个repli节点会替代成为leader节点，此时可能出现写入失败， # 当retris为0时，produce不会重复。retirs重发，此时repli节点完全成为leader节点，不会产生消息丢失。 spring.kafka.producer.retries=0 # 每次批量发送消息的数量,produce积累到一定数据，一次发送 #spring.kafka.producer.batch-size=16384 spring.kafka.producer.batch-size=16 # produce积累数据一次发送，缓存大小达到buffer.memory就发送数据 #spring.kafka.producer.buffer-memory=33554432 spring.kafka.producer.buffer-memory=128 #procedure要求leader在考虑完成请求之前收到的确认数，用于控制发送记录在服务端的持久化，其值可以为如下： #acks = 0 如果设置为零，则生产者将不会等待来自服务器的任何确认，该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下，无法保证服务器已收到记录，并且重试配置将不会生效（因为客户端通常不会知道任何故障），为每条记录返回的偏移量始终设置为-1。 #acks = 1 这意味着leader会将记录写入其本地日志，但无需等待所有副本服务器的完全确认即可做出回应，在这种情况下，如果leader在确认记录后立即失败，但在将数据复制到所有的副本服务器之前，则记录将会丢失。 #acks = all 这意味着leader将等待完整的同步副本集以确认记录，这保证了只要至少一个同步副本服务器仍然存活，记录就不会丢失，这是最强有力的保证，这相当于acks = -1的设置。 #可以设置的值为：all, -1, 0, 1 spring.kafka.producer.acks=1 # 指定消息key和消息体的编解码方式 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer #=============== consumer ======================= # 指定默认消费者group id --> 由于在kafka中，同一组中的consumer不会读取到同一个消息，依靠groud.id设置组名 spring.kafka.consumer.group-id=om.kafkaStudy # smallest和largest才有效，如果smallest重新0开始读取，如果是largest从logfile的offset读取。一般情况下我们都是设置smallest spring.kafka.consumer.auto-offset-reset=earliest # enable.auto.commit:true --> 设置自动提交offset spring.kafka.consumer.enable-auto-commit=true #如果‘enable.auto.commit‘为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。 spring.kafka.consumer.auto-commit-interval=100 # 指定消息key和消息体的编解码方式 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
package com.kafkastudy.kafka01.com.kafkastudy.kafka01;
import com.alibaba.fastjson.JSON;
import com.kafkastudy.kafka01.bean.User;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
@Component
public class ConsumerListener {
@KafkaListener(topics = "userTest")
public void onMessage(String user){
//insertIntoDb(buffer);//这里为插入数据库代码
System.out.println("------receive----------");
System.out.println(user);
System.out.println("------receive end----------");
}
}
package com.kafkastudy.kafka01.producer;
import com.alibaba.fastjson.JSON;
import com.kafkastudy.kafka01.bean.User;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
@Configuration
public class UserProducer {
@Autowired
private KafkaTemplate kafkaTemplate;
/**
* 发送数据
* @param userid
*/
public void sendData(long userid){
User userLog = new User();
userLog.setName("test");
userLog.setId(userid);
System.err.println("发送用户日志数据:"+userLog);
kafkaTemplate.send("userTest", JSON.toJSONString(userLog));
}
}
package com.kafkastudy.kafka01.bean;
import java.io.Serializable;
public class User implements Serializable {
private Long id;
private String name;
private Integer age;
/**
* transient 关键字修饰的字段不会被序列化
*/
private transient String desc;
public Long getId() {
return id;
}
public void setId(Long id) {
this.id = id;
}
public String getName() {
return name;
}
public void setName(String name) {
this.name = name;
}
public Integer getAge() {
return age;
}
public void setAge(Integer age) {
this.age = age;
}
public String getDesc() {
return desc;
}
public void setDesc(String desc) {
this.desc = desc;
}
@Override
public String toString() {
return "User{" +
"id=" + id +
", name=‘" + name + ‘\‘‘ +
", age=" + age +
", desc=‘" + desc + ‘\‘‘ +
‘}‘;
}
}
原文地址：https://www.cnblogs.com/alexgl2008/p/12158684.html
时间： 2024-10-05 21:47:13