diff --git a/pom.xml b/pom.xml
index dc9be7a..225a8dc 100644
--- a/pom.xml
+++ b/pom.xml
@@ -75,6 +75,11 @@
+
+ org.springframework.kafka
+ spring-kafka
+
+
junit
junit
diff --git a/src/main/java/cn/com/tenlion/usercenter/kafka/KafKaPublishConsumer.java b/src/main/java/cn/com/tenlion/usercenter/kafka/KafKaPublishConsumer.java
new file mode 100644
index 0000000..c4da093
--- /dev/null
+++ b/src/main/java/cn/com/tenlion/usercenter/kafka/KafKaPublishConsumer.java
@@ -0,0 +1,57 @@
+package cn.com.tenlion.usercenter.kafka;
+
+import cn.com.tenlion.usercenter.service.teammembergrid.ITeamMemberGridService;
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.JSONObject;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.ApplicationEventPublisher;
+import org.springframework.context.ApplicationEventPublisherAware;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @ClassName: KafKaPublishConsumer
+ * @Description: kafka消费者
+ * @Author: wanggeng
+ * @Date: 2021/12/23 10:06 AM
+ * @Version: 1.0
+ */
+@Component
+public class KafKaPublishConsumer implements ApplicationEventPublisherAware {
+
+ private ApplicationEventPublisher applicationEventPublisher;
+
+ @Autowired
+ private ITeamMemberGridService teamMemberGridService;
+ private MongoTemplate mongoTemplate;
+
+ public KafKaPublishConsumer(MongoTemplate mongoTemplate) {
+ this.mongoTemplate = mongoTemplate;
+ }
+
+ @KafkaListener(topics = {"C0021"})
+ public void gridMember(ConsumerRecord, ?> record) {
+ JSONObject publishObject = JSON.parseObject(record.value().toString());
+ System.out.println("收到消息:" + publishObject.toJSONString());
+ }
+
+ @KafkaListener(topics = {"C0022"})
+ public void mapGrid(ConsumerRecord, ?> record) {
+ JSONObject publishObject = JSON.parseObject(record.value().toString());
+ System.out.println("收到消息:" + publishObject.toJSONString());
+ }
+
+ @KafkaListener(topics = {"C0023"})
+ public void mapGridPoint(ConsumerRecord, ?> record) {
+ JSONObject publishObject = JSON.parseObject(record.value().toString());
+ System.out.println("收到消息:" + publishObject.toJSONString());
+ }
+
+ @Override
+ public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+ this.applicationEventPublisher = applicationEventPublisher;
+ }
+
+}
diff --git a/src/main/java/cn/com/tenlion/usercenter/pojo/dtos/kafka/SyncDataDTO.java b/src/main/java/cn/com/tenlion/usercenter/pojo/dtos/kafka/SyncDataDTO.java
new file mode 100644
index 0000000..bcabb89
--- /dev/null
+++ b/src/main/java/cn/com/tenlion/usercenter/pojo/dtos/kafka/SyncDataDTO.java
@@ -0,0 +1,17 @@
+package cn.com.tenlion.usercenter.pojo.dtos.kafka;
+
+/**
+ * @ClassName: SyncTableDTO
+ * @Description:
+ * @Author: wanggeng
+ * @Date: 2021/12/23 12:34 PM
+ * @Version: 1.0
+ */
+public class SyncDataDTO {
+
+ private String uid;
+ private String tableNumber;
+ private String action;
+ private String
+
+}
diff --git a/src/main/java/cn/com/tenlion/usercenter/service/grid/impl/GridSaveAfterHandlerImpl.java b/src/main/java/cn/com/tenlion/usercenter/service/grid/impl/GridSaveAfterHandlerImpl.java
new file mode 100644
index 0000000..b32982b
--- /dev/null
+++ b/src/main/java/cn/com/tenlion/usercenter/service/grid/impl/GridSaveAfterHandlerImpl.java
@@ -0,0 +1,30 @@
+package cn.com.tenlion.usercenter.service.grid.impl;
+
+import ink.wgink.interfaces.map.IGridSaveAfterHandler;
+import org.json.JSONObject;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.stereotype.Service;
+
+/**
+ * @ClassName: GridSaveAfterHandlerImpl
+ * @Description: 网格新增
+ * @Author: wanggeng
+ * @Date: 2021/12/23 11:27 AM
+ * @Version: 1.0
+ */
+@Service
+public class GridSaveAfterHandlerImpl implements IGridSaveAfterHandler {
+
+ @Autowired
+ private KafkaTemplate kafkaTemplate;
+
+ @Override
+ public void handle(String gridId, String gridCode, String gridName, String fillColor, String areaCode, String areaName) {
+ JSONObject gridJSONObject = new JSONObject();
+ gridJSONObject.put("uid", gridId);
+ gridJSONObject.put("action", "save");
+
+ }
+
+}
diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml
index a1d3fb6..4408680 100644
--- a/src/main/resources/application-dev.yml
+++ b/src/main/resources/application-dev.yml
@@ -67,6 +67,27 @@ spring:
max-wait: 1ms
max-idle: 8
min-idle: 0
+ kafka:
+ bootstrap-servers: 192.168.0.156:9093
+ producer:
+ # 写入失败时,重试次数。当leader节点失效,一个repli节点会替代成为leader节点,此时可能出现写入失败,
+ # 当retris为0时,produce不会重复。retirs重发,此时repli节点完全成为leader节点,不会产生消息丢失。
+ retries: 0
+ #procedure要求leader在考虑完成请求之前收到的确认数,用于控制发送记录在服务端的持久化,其值可以为如下:
+ #acks = 0 如果设置为零,则生产者将不会等待来自服务器的任何确认,该记录将立即添加到套接字缓冲区并视为已发送。在这种情况下,无法保证服务器已收到记录,并且重试配置将不会生效(因为客户端通常不会知道任何故障),为每条记录返回的偏移量始终设置为-1。
+ #acks = 1 这意味着leader会将记录写入其本地日志,但无需等待所有副本服务器的完全确认即可做出回应,在这种情况下,如果leader在确认记录后立即失败,但在将数据复制到所有的副本服务器之前,则记录将会丢失。
+ #acks = all 这意味着leader将等待完整的同步副本集以确认记录,这保证了只要至少一个同步副本服务器仍然存活,记录就不会丢失,这是最强有力的保证,这相当于acks = -1的设置。
+ #可以设置的值为:all, -1, 0, 1
+ acks: 1
+ consumer:
+ group-id: SmartCity
+ # smallest和largest才有效,如果smallest重新0开始读取,如果是largest从logfile的offset读取。一般情况下我们都是设置smallest
+ auto-offset-reset: earliest
+ # 设置自动提交offset
+ enable-auto-commit: true
+ # 如果'enable.auto.commit'为true,则消费者偏移自动提交给Kafka的频率(以毫秒为单位),默认值为5000。
+ auto-commit-interval: 100
+ max-poll-records: 5
mybatis:
config-location: classpath:mybatis/mybatis-config.xml