fix(): 事件基础修改,kafka注入
parent
c35dea9159
commit
cc99d65102
|
|
@ -0,0 +1,49 @@
|
||||||
|
package com.muyu.common.core.constant;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* kafka常量信息
|
||||||
|
* @program: cloud-server
|
||||||
|
* @author: cuiyongxing
|
||||||
|
* @create: 2024-09-28 12:18
|
||||||
|
**/
|
||||||
|
public class KafkaConstant {
|
||||||
|
|
||||||
|
public static final String BOOTSTRAP_SERVERS = "bootstrap.servers";
|
||||||
|
|
||||||
|
public static final String RETRIES = "retries";
|
||||||
|
|
||||||
|
public static final String ACKS = "acks";
|
||||||
|
|
||||||
|
public static final String BATCH_SIZE = "batch.size";
|
||||||
|
|
||||||
|
public static final String BUFFER_MEMORY = "buffer-memory";
|
||||||
|
|
||||||
|
public static final String KEY_SERIALIZER = "key.serializer";
|
||||||
|
|
||||||
|
public static final String VALUE_SERIALIZER = "value.serializer";
|
||||||
|
|
||||||
|
public static final String ENABLE_AUTO_COMMIT = "enable.auto.commit";
|
||||||
|
|
||||||
|
public static final String AUTO_COMMIT_INTERVAL = "auto.commit.interval.ms";
|
||||||
|
|
||||||
|
public static final String AUTO_OFFSET_RESET = "auto.offset.reset";
|
||||||
|
|
||||||
|
public static final String FETCH_MAX_WAIT = "fetch.max.wait";
|
||||||
|
|
||||||
|
public static final String FETCH_MIN_SIZE = "fetch.min.size";
|
||||||
|
|
||||||
|
public static final String HEARTBEAT_INTERVAL = "heartbeat.interval";
|
||||||
|
|
||||||
|
public static final String MAX_POLL_RECORDS = "max.poll.records";
|
||||||
|
|
||||||
|
public static final String KEY_DESERIALIZER = "key.deserializer";
|
||||||
|
|
||||||
|
public static final String VALUE_DESERIALIZER = "value.deserializer";
|
||||||
|
|
||||||
|
public static final String TOPIC = "topic";
|
||||||
|
|
||||||
|
public static final String GROUP_ID = "group.id";
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,40 @@
|
||||||
|
<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||||
|
<modelVersion>4.0.0</modelVersion>
|
||||||
|
<parent>
|
||||||
|
<groupId>com.muyu</groupId>
|
||||||
|
<artifactId>cloud-common</artifactId>
|
||||||
|
<version>3.6.3</version>
|
||||||
|
</parent>
|
||||||
|
|
||||||
|
<artifactId>cloud-common-kafka</artifactId>
|
||||||
|
<description>
|
||||||
|
cloud-common-kafka消息队列
|
||||||
|
</description>
|
||||||
|
<properties>
|
||||||
|
<maven.compiler.source>17</maven.compiler.source>
|
||||||
|
<maven.compiler.target>17</maven.compiler.target>
|
||||||
|
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||||
|
</properties>
|
||||||
|
<dependencies>
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.apache.kafka</groupId>
|
||||||
|
<artifactId>kafka-clients</artifactId>
|
||||||
|
<version>3.0.0</version>
|
||||||
|
</dependency>
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.muyu</groupId>
|
||||||
|
<artifactId>cloud-common-core</artifactId>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
<dependency>
|
||||||
|
<groupId>org.apache.kafka</groupId>
|
||||||
|
<artifactId>kafka-clients</artifactId>
|
||||||
|
<version>2.8.0</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
|
|
||||||
|
</dependencies>
|
||||||
|
</project>
|
||||||
|
|
@ -0,0 +1,96 @@
|
||||||
|
package com.muyu.common.kafka.config;
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
import com.muyu.common.core.constant.KafkaConstant;
|
||||||
|
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||||
|
import org.apache.kafka.common.serialization.StringDeserializer;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* kafka消费者配置类
|
||||||
|
* @program: cloud-server
|
||||||
|
* @author: cuiyongxing
|
||||||
|
* @create: 2024-09-28 14:28
|
||||||
|
**/
|
||||||
|
@Configuration
|
||||||
|
public class KafkaConsumerConfig {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 服务端id+端口号
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.bootstrap-servers}")
|
||||||
|
private String bootstrapServers;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 偏移量
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.enable-auto-commit}")
|
||||||
|
private Boolean enableAutoCommit;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 自动提交时间间隔
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.auto-commit-interval}")
|
||||||
|
private Integer autoCommitInterval;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 偏移量配置
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.auto-offset-reset}")
|
||||||
|
private String autoOffsetReset;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 阻塞最大时间
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.fetch-max-wait}")
|
||||||
|
private Integer fetchMaxWait;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 请求最小字节
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.fetch-min-size}")
|
||||||
|
private Integer fetchMinSize;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 心跳间隔时间
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.heartbeat-interval}")
|
||||||
|
private Integer heartbeatInterval;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 最大记录条数
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.max-poll-records}")
|
||||||
|
private Integer maxPollRecords;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 消费组
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.consumer.group-id}")
|
||||||
|
private String groupId;
|
||||||
|
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public KafkaConsumer kafkaConsumer(){
|
||||||
|
Map<String,Object> configs = new HashMap<>();
|
||||||
|
configs.put(KafkaConstant.BOOTSTRAP_SERVERS, bootstrapServers);
|
||||||
|
configs.put(KafkaConstant.ENABLE_AUTO_COMMIT, enableAutoCommit);
|
||||||
|
configs.put(KafkaConstant.AUTO_COMMIT_INTERVAL, autoCommitInterval);
|
||||||
|
configs.put(KafkaConstant.AUTO_OFFSET_RESET, autoOffsetReset);
|
||||||
|
configs.put(KafkaConstant.FETCH_MAX_WAIT, fetchMaxWait);
|
||||||
|
configs.put(KafkaConstant.FETCH_MIN_SIZE, fetchMinSize);
|
||||||
|
configs.put(KafkaConstant.HEARTBEAT_INTERVAL, heartbeatInterval);
|
||||||
|
configs.put(KafkaConstant.MAX_POLL_RECORDS, maxPollRecords);
|
||||||
|
configs.put(KafkaConstant.GROUP_ID, groupId);
|
||||||
|
StringDeserializer keyDeserializer = new StringDeserializer();
|
||||||
|
StringDeserializer valueDeserializer = new StringDeserializer();
|
||||||
|
return new KafkaConsumer(configs, keyDeserializer, valueDeserializer);
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,68 @@
|
||||||
|
package com.muyu.common.kafka.config;
|
||||||
|
|
||||||
|
import com.muyu.common.core.constant.KafkaConstant;
|
||||||
|
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||||
|
import org.apache.kafka.common.serialization.StringSerializer;
|
||||||
|
import org.springframework.beans.factory.annotation.Value;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
|
||||||
|
|
||||||
|
import java.util.HashMap;
|
||||||
|
import java.util.Map;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* kafka生产者配置信息
|
||||||
|
* @program: cloud-server
|
||||||
|
* @author: cuiyongxing
|
||||||
|
* @create: 2024-09-28 12:03
|
||||||
|
**/
|
||||||
|
@Configuration
|
||||||
|
public class KafkaProducerConfig {
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 服务端ip+端口号
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.producer.bootstrap-servers}")
|
||||||
|
private String bootstrapServers;
|
||||||
|
/**
|
||||||
|
* 重试次数
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.producer.retries}")
|
||||||
|
private Integer retries;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 默认批量大小
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.producer.batch-size}")
|
||||||
|
private Integer batchSize;
|
||||||
|
/**
|
||||||
|
* 总内存字节数
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.producer.buffer-memory}")
|
||||||
|
private Integer bufferMemory;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 偏移量
|
||||||
|
*/
|
||||||
|
@Value("${spring.kafka.producer.acks}")
|
||||||
|
private String acks;
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public KafkaProducer kafkaProducer() {
|
||||||
|
Map<String, Object> configs = new HashMap<>();
|
||||||
|
configs.put(KafkaConstant.BOOTSTRAP_SERVERS, bootstrapServers);
|
||||||
|
configs.put(KafkaConstant.RETRIES, retries);
|
||||||
|
configs.put(KafkaConstant.BATCH_SIZE, batchSize);
|
||||||
|
configs.put(KafkaConstant.BUFFER_MEMORY, bufferMemory);
|
||||||
|
configs.put(KafkaConstant.ACKS, acks);
|
||||||
|
StringSerializer keySerializer = new StringSerializer();
|
||||||
|
StringSerializer valueSerializer = new StringSerializer();
|
||||||
|
KafkaProducer kafkaProducer = new KafkaProducer<>(configs, keySerializer, valueSerializer);
|
||||||
|
return kafkaProducer;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,2 @@
|
||||||
|
com.muyu.common.kafka.config.KafkaConsumerConfig
|
||||||
|
com.muyu.common.kafka.config.KafkaProducerConfig
|
||||||
|
|
@ -22,6 +22,7 @@
|
||||||
<module>cloud-common-rabbit</module>
|
<module>cloud-common-rabbit</module>
|
||||||
<module>cloud-common-saas</module>
|
<module>cloud-common-saas</module>
|
||||||
<module>cloud-common-swagger</module>
|
<module>cloud-common-swagger</module>
|
||||||
|
<module>cloud-common-kafka</module>
|
||||||
</modules>
|
</modules>
|
||||||
|
|
||||||
<artifactId>cloud-common</artifactId>
|
<artifactId>cloud-common</artifactId>
|
||||||
|
|
|
||||||
|
|
@ -110,11 +110,18 @@
|
||||||
<artifactId>node-commons</artifactId>
|
<artifactId>node-commons</artifactId>
|
||||||
<version>1.3.2</version>
|
<version>1.3.2</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
<dependency>
|
<dependency>
|
||||||
<groupId>org.springframework.kafka</groupId>
|
<groupId>org.apache.kafka</groupId>
|
||||||
<artifactId>spring-kafka</artifactId>
|
<artifactId>kafka-clients</artifactId>
|
||||||
|
<version>2.8.0</version>
|
||||||
</dependency>
|
</dependency>
|
||||||
|
|
||||||
|
<dependency>
|
||||||
|
<groupId>com.muyu</groupId>
|
||||||
|
<artifactId>cloud-common-kafka</artifactId>
|
||||||
|
<version>3.6.3</version>
|
||||||
|
</dependency>
|
||||||
|
|
||||||
</dependencies>
|
</dependencies>
|
||||||
|
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,26 @@
|
||||||
|
package com.muyu.event.basic;
|
||||||
|
|
||||||
|
import com.alibaba.fastjson2.JSONObject;
|
||||||
|
import org.springframework.context.ApplicationEvent;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 事件类型
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.Basic
|
||||||
|
* @name：EventCustom
|
||||||
|
* @date：2024/9/29 21:17
|
||||||
|
*/
|
||||||
|
public class EventCustom extends ApplicationEvent {
|
||||||
|
|
||||||
|
private JSONObject data;
|
||||||
|
|
||||||
|
public EventCustom(Object source,JSONObject data) {
|
||||||
|
super(source);
|
||||||
|
this.data=data;
|
||||||
|
}
|
||||||
|
|
||||||
|
public JSONObject getData(){
|
||||||
|
return data;
|
||||||
|
}
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,17 @@
|
||||||
|
package com.muyu.event.basic;
|
||||||
|
|
||||||
|
|
||||||
|
import org.springframework.context.ApplicationListener;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 事件监听接口
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.basic
|
||||||
|
* @name：EventListener
|
||||||
|
* @date：2024/9/29 21:21
|
||||||
|
*/
|
||||||
|
public interface EventListener extends ApplicationListener<EventCustom> {
|
||||||
|
|
||||||
|
void onEvent(EventCustom event);
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,33 @@
|
||||||
|
package com.muyu.event.basic;
|
||||||
|
|
||||||
|
import com.alibaba.fastjson2.JSONB;
|
||||||
|
import com.alibaba.fastjson2.JSONObject;
|
||||||
|
import org.springframework.context.ApplicationEventPublisher;
|
||||||
|
import org.springframework.context.ApplicationEventPublisherAware;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 策略发送事件
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.basic
|
||||||
|
* @name：EventPublisher
|
||||||
|
* @date：2024/9/29 22:01
|
||||||
|
*/
|
||||||
|
@Component
|
||||||
|
public class EventPublisher implements ApplicationEventPublisherAware {
|
||||||
|
|
||||||
|
private ApplicationEventPublisher publisher;
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
|
||||||
|
this.publisher=applicationEventPublisher;
|
||||||
|
}
|
||||||
|
|
||||||
|
public void publishEvent(JSONObject jsonObject){
|
||||||
|
EventCustom event = new EventCustom(this, jsonObject);
|
||||||
|
publisher.publishEvent(event);
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -1,25 +0,0 @@
|
||||||
package com.muyu.event.basics;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @author 刘武
|
|
||||||
* @package：
|
|
||||||
* @name：EventHandler
|
|
||||||
* @date：2024/9/29
|
|
||||||
*/
|
|
||||||
public class EventHandler {
|
|
||||||
|
|
||||||
private static final ThreadLocal<EventQueueConfig> EVENT_THREAD = new ThreadLocal<>();
|
|
||||||
|
|
||||||
public static void set(final EventQueueConfig handler) {
|
|
||||||
EVENT_THREAD.set(handler);
|
|
||||||
}
|
|
||||||
|
|
||||||
public static EventQueueConfig get() {
|
|
||||||
return EVENT_THREAD.get();
|
|
||||||
}
|
|
||||||
|
|
||||||
public static void remove(){
|
|
||||||
EVENT_THREAD.remove();
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,25 +0,0 @@
|
||||||
package com.muyu.event.basics;
|
|
||||||
|
|
||||||
|
|
||||||
public abstract class EventProcessBasics {
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 下一个事件对象
|
|
||||||
*/
|
|
||||||
protected EventProcessBasics nextEvent;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 下一个事件
|
|
||||||
* @param nextHandler 下一个事件处理
|
|
||||||
*/
|
|
||||||
public void setNextHandler(EventProcessBasics nextHandler) {
|
|
||||||
this.nextEvent = nextHandler;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 事件处理抽象类
|
|
||||||
* @param eventKey 事件唯一key
|
|
||||||
*/
|
|
||||||
public abstract void handleEvent(String eventKey);
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,35 +0,0 @@
|
||||||
package com.muyu.event.basics;
|
|
||||||
|
|
||||||
import lombok.AllArgsConstructor;
|
|
||||||
import lombok.Builder;
|
|
||||||
import lombok.Data;
|
|
||||||
import lombok.NoArgsConstructor;
|
|
||||||
|
|
||||||
import java.util.concurrent.LinkedBlockingDeque;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @author 刘武
|
|
||||||
* @package：
|
|
||||||
* @name：EventQueueConfig
|
|
||||||
* @date：2024/9/29
|
|
||||||
*/
|
|
||||||
@Data
|
|
||||||
@Builder
|
|
||||||
@AllArgsConstructor
|
|
||||||
@NoArgsConstructor
|
|
||||||
public class EventQueueConfig {
|
|
||||||
|
|
||||||
private LinkedBlockingDeque<EventProcessBasics> taskNodeQueue = new LinkedBlockingDeque<>();
|
|
||||||
|
|
||||||
public void addEvent(EventProcessBasics obj){
|
|
||||||
this.taskNodeQueue.add(obj);
|
|
||||||
}
|
|
||||||
|
|
||||||
public boolean hashEventNext(){
|
|
||||||
return !taskNodeQueue.isEmpty();
|
|
||||||
}
|
|
||||||
|
|
||||||
private EventProcessBasics nextTaskNode(){
|
|
||||||
return taskNodeQueue.poll();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -1,28 +0,0 @@
|
||||||
package com.muyu.event.basics;
|
|
||||||
|
|
||||||
|
|
||||||
import com.muyu.event.domian.EventActuate;
|
|
||||||
import org.springframework.context.ApplicationEvent;
|
|
||||||
|
|
||||||
import java.util.List;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @author 刘武
|
|
||||||
* @package：
|
|
||||||
* @name：StartEvent
|
|
||||||
* @date：2024/9/29
|
|
||||||
*/
|
|
||||||
|
|
||||||
public class StartEvent extends ApplicationEvent {
|
|
||||||
|
|
||||||
private EventActuate eventActuate;
|
|
||||||
|
|
||||||
public StartEvent(EventActuate source) {
|
|
||||||
super(source);
|
|
||||||
this.eventActuate = source;
|
|
||||||
}
|
|
||||||
|
|
||||||
public EventActuate getEventActuate() {
|
|
||||||
return eventActuate;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -0,0 +1,23 @@
|
||||||
|
package com.muyu.event.config;
|
||||||
|
|
||||||
|
|
||||||
|
import com.muyu.event.listener.AddDatabaseListener;
|
||||||
|
import org.springframework.context.annotation.Bean;
|
||||||
|
import org.springframework.context.annotation.Configuration;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.config
|
||||||
|
* @name：EventConfig
|
||||||
|
* @date：2024/9/29 21:13
|
||||||
|
*/
|
||||||
|
@Configuration
|
||||||
|
public class EventConfig {
|
||||||
|
|
||||||
|
@Bean
|
||||||
|
public AddDatabaseListener addDatabaseListener() {
|
||||||
|
return new AddDatabaseListener();
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -1,129 +0,0 @@
|
||||||
package com.muyu.event.config;
|
|
||||||
|
|
||||||
|
|
||||||
import org.apache.kafka.clients.consumer.ConsumerConfig;
|
|
||||||
import org.springframework.beans.factory.annotation.Value;
|
|
||||||
import org.springframework.boot.SpringBootConfiguration;
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
|
|
||||||
import org.springframework.kafka.config.KafkaListenerContainerFactory;
|
|
||||||
import org.springframework.kafka.core.ConsumerFactory;
|
|
||||||
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
|
|
||||||
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
|
|
||||||
import org.springframework.kafka.listener.ContainerProperties;
|
|
||||||
import org.springframework.kafka.support.serializer.JsonDeserializer;
|
|
||||||
|
|
||||||
|
|
||||||
import java.util.HashMap;
|
|
||||||
import java.util.Map;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @author 徐一杰
|
|
||||||
* @date 2022/10/31 18:05
|
|
||||||
* kafka配置，也可以写在yml，这个文件会覆盖yml
|
|
||||||
*/
|
|
||||||
@SpringBootConfiguration
|
|
||||||
public class KafkaConsumerConfig {
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 配置 Kafka的 主机地址
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.consumer.bootstrap-servers}")
|
|
||||||
private String bootstrapServers;
|
|
||||||
/**
|
|
||||||
* 配置分分组
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.consumer.group-id}")
|
|
||||||
private String groupId;
|
|
||||||
/**
|
|
||||||
* 是否自动提交 偏移量
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.consumer.enable-auto-commit}")
|
|
||||||
private boolean enableAutoCommit;
|
|
||||||
/**
|
|
||||||
* 消费者与Kafka的心跳续约的会话超时时间
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.properties.session.timeout.ms}")
|
|
||||||
private String sessionTimeout;
|
|
||||||
/**
|
|
||||||
* 两次poll之间的最大间隔，默认值为5分钟。如果超过这个间隔会触发reBalance
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.properties.max.poll.interval.ms}")
|
|
||||||
private String maxPollIntervalTime;
|
|
||||||
|
|
||||||
@Value("${spring.kafka.consumer.max-poll-records}")
|
|
||||||
private String maxPollRecords;
|
|
||||||
|
|
||||||
@Value("${spring.kafka.consumer.auto-offset-reset}")
|
|
||||||
private String autoOffsetReset;
|
|
||||||
|
|
||||||
@Value("${spring.kafka.listener.concurrency}")
|
|
||||||
private Integer concurrency;
|
|
||||||
|
|
||||||
@Value("${spring.kafka.listener.missing-topics-fatal}")
|
|
||||||
private boolean missingTopicsFatal;
|
|
||||||
|
|
||||||
@Value("${spring.kafka.listener.poll-timeout}")
|
|
||||||
private long pollTimeout;
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public Map<String, Object> consumerConfigs() {
|
|
||||||
Map<String, Object> propsMap = new HashMap<>(16);
|
|
||||||
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
|
||||||
propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
|
|
||||||
//是否自动提交偏移量，默认值是true，为了避免出现重复数据和数据丢失，可以把它设置为false，然后手动提交偏移量
|
|
||||||
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
|
|
||||||
//自动提交的时间间隔，自动提交开启时生效
|
|
||||||
propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "2000");
|
|
||||||
//该属性指定了消费者在读取一个没有偏移量的分区或者偏移量无效的情况下该作何处理：
|
|
||||||
//earliest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，从头开始消费分区的记录
|
|
||||||
//latest：当各分区下有已提交的offset时，从提交的offset开始消费；无提交的offset时，消费新产生的该分区下的数据（在消费者启动之后生成的记录）
|
|
||||||
//none：当各分区都存在已提交的offset时，从提交的offset开始消费；只要有一个分区不存在已提交的offset，则抛出异常
|
|
||||||
propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
|
|
||||||
//两次poll之间的最大间隔，默认值为5分钟。如果超过这个间隔会触发reBalance
|
|
||||||
propsMap.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, maxPollIntervalTime);
|
|
||||||
//这个参数定义了poll方法最多可以拉取多少条消息，默认值为500。如果在拉取消息的时候新消息不足500条，那有多少返回多少；如果超过500条，每次只返回500。
|
|
||||||
//这个默认值在有些场景下太大，有些场景很难保证能够在5min内处理完500条消息，
|
|
||||||
//如果消费者无法在5分钟内处理完500条消息的话就会触发reBalance,
|
|
||||||
//然后这批消息会被分配到另一个消费者中，还是会处理不完，这样这批消息就永远也处理不完。
|
|
||||||
//要避免出现上述问题，提前评估好处理一条消息最长需要多少时间，然后覆盖默认的max.poll.records参数
|
|
||||||
//注：需要开启BatchListener批量监听才会生效，如果不开启BatchListener则不会出现reBalance情况
|
|
||||||
propsMap.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, maxPollRecords);
|
|
||||||
//当broker多久没有收到consumer的心跳请求后就触发reBalance，默认值是10s
|
|
||||||
propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
|
|
||||||
//序列化（建议使用Json，这种序列化方式可以无需额外配置传输实体类）
|
|
||||||
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
|
|
||||||
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
|
|
||||||
return propsMap;
|
|
||||||
}
|
|
||||||
|
|
||||||
@Bean
|
|
||||||
public ConsumerFactory<Object, Object> consumerFactory() {
|
|
||||||
// 配置消费者的 Json 反序列化的可信赖包，反序列化实体类需要
|
|
||||||
try (JsonDeserializer<Object> deserializer = new JsonDeserializer<>()) {
|
|
||||||
deserializer.trustedPackages("*");
|
|
||||||
return new DefaultKafkaConsumerFactory<>(consumerConfigs(), new JsonDeserializer<>(), deserializer);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* kafka监听容器工厂 负责 从 Kafka的主题中 取出消息进行消费 可以设置消费者的配置
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
@Bean
|
|
||||||
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Object, Object>> kafkaListenerContainerFactory() {
|
|
||||||
ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
|
|
||||||
factory.setConsumerFactory(consumerFactory());
|
|
||||||
//在侦听器容器中运行的线程数，一般设置为 机器数*分区数
|
|
||||||
factory.setConcurrency(concurrency);
|
|
||||||
// 消费监听接口监听的主题不存在时，默认会报错，所以设置为false忽略错误
|
|
||||||
factory.setMissingTopicsFatal(missingTopicsFatal);
|
|
||||||
// 自动提交关闭，需要设置手动消息确认
|
|
||||||
factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
|
|
||||||
factory.getContainerProperties().setPollTimeout(pollTimeout);
|
|
||||||
// 设置为批量监听，需要用List接收
|
|
||||||
// factory.setBatchListener(true);
|
|
||||||
return factory;
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,127 +0,0 @@
|
||||||
package com.muyu.event.config;
|
|
||||||
|
|
||||||
import org.apache.kafka.clients.producer.ProducerConfig;
|
|
||||||
import org.springframework.beans.factory.annotation.Value;
|
|
||||||
import org.springframework.context.annotation.Bean;
|
|
||||||
import org.springframework.context.annotation.Configuration;
|
|
||||||
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
|
|
||||||
import org.springframework.kafka.core.KafkaTemplate;
|
|
||||||
import org.springframework.kafka.core.ProducerFactory;
|
|
||||||
import org.springframework.kafka.transaction.KafkaTransactionManager;
|
|
||||||
|
|
||||||
import java.util.HashMap;
|
|
||||||
import java.util.Map;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 主题生产者的配置类
|
|
||||||
*/
|
|
||||||
@Configuration
|
|
||||||
public class KafkaProviderConfig {
|
|
||||||
|
|
||||||
/**
|
|
||||||
* kafka 的主机地址
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.bootstrap-servers}")
|
|
||||||
private String bootstrapServers;
|
|
||||||
/**
|
|
||||||
* 配置 Kafka的事务
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.transaction-id-prefix}")
|
|
||||||
private String transactionIdPrefix;
|
|
||||||
/**
|
|
||||||
* 发送确认机制
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.acks}")
|
|
||||||
private String acks;
|
|
||||||
/**
|
|
||||||
* 发送重试
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.retries}")
|
|
||||||
private String retries;
|
|
||||||
/**
|
|
||||||
* 发送消息的批次大小
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.batch-size}")
|
|
||||||
private String batchSize;
|
|
||||||
/**
|
|
||||||
* 消息的缓冲区内存大小
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.buffer-memory}")
|
|
||||||
private String bufferMemory;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 设置 健的序列化方式
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.key-serializer}")
|
|
||||||
private String keySerializer;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 设置 值的序列化方式
|
|
||||||
*/
|
|
||||||
@Value("${spring.kafka.producer.value-serializer}")
|
|
||||||
private String valueSerializer;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 构建 map 配置消息生产者对象的配置
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
@Bean
|
|
||||||
public Map<String, Object> producerConfigs() {
|
|
||||||
Map<String, Object> props = new HashMap<>(16);
|
|
||||||
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
|
|
||||||
//acks=0 ： 生产者在成功写入消息之前不会等待任何来自服务器的响应。
|
|
||||||
//acks=1 ： 只要集群的首领节点收到消息，生产者就会收到一个来自服务器成功响应。
|
|
||||||
//acks=all ：只有当所有参与复制的节点全部收到消息时，生产者才会收到一个来自服务器的成功响应。
|
|
||||||
//开启事务必须设为all
|
|
||||||
props.put(ProducerConfig.ACKS_CONFIG, acks);
|
|
||||||
//发生错误后，消息重发的次数，开启事务必须大于0
|
|
||||||
props.put(ProducerConfig.RETRIES_CONFIG, retries);
|
|
||||||
//当多个消息发送到相同分区时,生产者会将消息打包到一起,以减少请求交互. 而不是一条条发送
|
|
||||||
//批次的大小可以通过batch.size 参数设置.默认是16KB
|
|
||||||
//较小的批次大小有可能降低吞吐量（批次大小为0则完全禁用批处理）。
|
|
||||||
//比如说，kafka里的消息5秒钟Batch才凑满了16KB，才能发送出去。那这些消息的延迟就是5秒钟
|
|
||||||
//实测batchSize这个参数没有用
|
|
||||||
props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
|
|
||||||
//有的时刻消息比较少,过了很久,比如5min也没有凑够16KB,这样延时就很大,所以需要一个参数. 再设置一个时间,到了这个时间,
|
|
||||||
//即使数据没达到16KB,也将这个批次发送出去
|
|
||||||
props.put(ProducerConfig.LINGER_MS_CONFIG, "5000");
|
|
||||||
//生产者内存缓冲区的大小
|
|
||||||
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
|
|
||||||
//反序列化，和生产者的序列化方式对应
|
|
||||||
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
|
|
||||||
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);
|
|
||||||
return props;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 构建 主题生产者工厂
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
@Bean
|
|
||||||
public ProducerFactory<Object, Object> producerFactory() {
|
|
||||||
DefaultKafkaProducerFactory<Object, Object> factory = new DefaultKafkaProducerFactory<>(producerConfigs());
|
|
||||||
//开启事务，会导致 LINGER_MS_CONFIG 配置失效
|
|
||||||
factory.setTransactionIdPrefix(transactionIdPrefix);
|
|
||||||
return factory;
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 配置 Kafka的事务管理器
|
|
||||||
* @param producerFactory
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
@Bean
|
|
||||||
public KafkaTransactionManager<Object, Object> kafkaTransactionManager(ProducerFactory<Object, Object> producerFactory) {
|
|
||||||
return new KafkaTransactionManager<>(producerFactory);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 构建 KafkaTemplate
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
@Bean
|
|
||||||
public KafkaTemplate<Object, Object> kafkaTemplate() {
|
|
||||||
return new KafkaTemplate<>(producerFactory());
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,65 +0,0 @@
|
||||||
package com.muyu.event.config;
|
|
||||||
|
|
||||||
import jakarta.annotation.Nullable;
|
|
||||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
|
||||||
import org.apache.kafka.clients.producer.RecordMetadata;
|
|
||||||
import org.springframework.beans.factory.annotation.Autowired;
|
|
||||||
import org.springframework.kafka.core.KafkaTemplate;
|
|
||||||
import org.springframework.kafka.support.ProducerListener;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
import javax.annotation.PostConstruct;
|
|
||||||
|
|
||||||
@Component
|
|
||||||
public class KafkaSendResultHandler implements ProducerListener<Object,Object> {
|
|
||||||
|
|
||||||
@Autowired
|
|
||||||
private KafkaTemplate<Object,Object> kafkaTemplate;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* bean 初始化方法
|
|
||||||
*/
|
|
||||||
@PostConstruct
|
|
||||||
public void init(){
|
|
||||||
this.kafkaTemplate.setProducerListener(this);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 消息发送到Kafka成功的回调
|
|
||||||
* @param producerRecord
|
|
||||||
* @param recordMetadata
|
|
||||||
*/
|
|
||||||
@Override
|
|
||||||
public void onSuccess(ProducerRecord producerRecord, RecordMetadata recordMetadata){
|
|
||||||
System.out.println("信息发送成功："+ producerRecord.toString());
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 息发送到 Kafka 失败的回调
|
|
||||||
* @param producerRecord the failed record
|
|
||||||
* @param recordMetadata The metadata for the record that was sent (i.e. the partition
|
|
||||||
* and offset). If an error occurred, metadata will contain only valid topic and maybe
|
|
||||||
* the partition. If the partition is not provided in the ProducerRecord and an error
|
|
||||||
* occurs before partition is assigned, then the partition will be set to
|
|
||||||
* RecordMetadata.UNKNOWN_PARTITION.
|
|
||||||
* @param exception the exception thrown
|
|
||||||
*/
|
|
||||||
@Override
|
|
||||||
public void onError(ProducerRecord producerRecord, @Nullable RecordMetadata recordMetadata,
|
|
||||||
Exception exception){
|
|
||||||
System.out.println("消息发送失败: "+ producerRecord.toString());
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,34 +0,0 @@
|
||||||
package com.muyu.event.config;
|
|
||||||
|
|
||||||
|
|
||||||
import lombok.NonNull;
|
|
||||||
import org.apache.kafka.clients.consumer.Consumer;
|
|
||||||
import org.springframework.kafka.listener.KafkaListenerErrorHandler;
|
|
||||||
import org.springframework.kafka.listener.ListenerExecutionFailedException;
|
|
||||||
import org.springframework.messaging.Message;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
@Component
|
|
||||||
public class MyKafkaListenerErrorHandler implements KafkaListenerErrorHandler {
|
|
||||||
|
|
||||||
@Override
|
|
||||||
@NonNull
|
|
||||||
public Object handleError(@NonNull Message<?> message,
|
|
||||||
ListenerExecutionFailedException exception) {
|
|
||||||
return new Object();
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
@NonNull
|
|
||||||
public Object handleError(@NonNull Message<?> message,
|
|
||||||
@NonNull ListenerExecutionFailedException exception,
|
|
||||||
Consumer<?, ?> consumer) {
|
|
||||||
System.out.println("消息详情："+ message);
|
|
||||||
System.out.println("异常信息："+ exception);
|
|
||||||
System.out.println("消费者详情：" +consumer.groupMetadata());
|
|
||||||
System.out.println("监听主题："+ consumer.listTopics());
|
|
||||||
return KafkaListenerErrorHandler.super.handleError(message, exception, consumer);
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,15 +0,0 @@
|
||||||
package com.muyu.event.constant;
|
|
||||||
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 事件常量
|
|
||||||
* @author 刘武
|
|
||||||
* @package：com.muyu.event.constant
|
|
||||||
* @name：EventConstant
|
|
||||||
* @date：2024/9/28 19:25
|
|
||||||
*/
|
|
||||||
|
|
||||||
public interface EventConstant {
|
|
||||||
|
|
||||||
String STORAGE_EVENT = "storageEvent";
|
|
||||||
}
|
|
||||||
|
|
@ -1,31 +0,0 @@
|
||||||
package com.muyu.event.consumer;
|
|
||||||
|
|
||||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
|
||||||
import org.springframework.kafka.annotation.KafkaListener;
|
|
||||||
import org.springframework.kafka.support.Acknowledgment;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* kafka监听
|
|
||||||
* @author 刘武
|
|
||||||
* @package：com.muyu.event.consumer
|
|
||||||
* @name：KafkaConsumer
|
|
||||||
* @date：2024/9/28 23:34
|
|
||||||
*/
|
|
||||||
|
|
||||||
public class KafkaConsumer {
|
|
||||||
|
|
||||||
|
|
||||||
@KafkaListener(topics = "data")
|
|
||||||
public void dataKafkaConsumer(ConsumerRecord<Object,Object> consumerRecord, Acknowledgment acknowledgment){
|
|
||||||
Object key = consumerRecord.key();
|
|
||||||
Object value = consumerRecord.value();
|
|
||||||
|
|
||||||
//事件调用
|
|
||||||
|
|
||||||
|
|
||||||
//消息确认消费
|
|
||||||
acknowledgment.acknowledge();
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -0,0 +1,59 @@
|
||||||
|
package com.muyu.event.consumer;
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
import com.alibaba.fastjson2.JSONObject;
|
||||||
|
import com.muyu.event.basic.EventPublisher;
|
||||||
|
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||||
|
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
import org.springframework.boot.ApplicationArguments;
|
||||||
|
import org.springframework.boot.ApplicationRunner;
|
||||||
|
import org.springframework.stereotype.Component;
|
||||||
|
|
||||||
|
import java.time.Duration;
|
||||||
|
import java.util.Collections;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
import static org.bouncycastle.asn1.x500.style.RFC4519Style.l;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* kafka监听
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.consumer
|
||||||
|
* @name：KafkaConsumer
|
||||||
|
* @date：2024/9/28 23:34
|
||||||
|
*/
|
||||||
|
@Component
|
||||||
|
public class MessageConsumer implements ApplicationRunner {
|
||||||
|
|
||||||
|
@Autowired
|
||||||
|
public KafkaConsumer consumer;
|
||||||
|
@Autowired
|
||||||
|
private EventPublisher eventPublisher;
|
||||||
|
|
||||||
|
private final String topic="vehicle";
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void run(ApplicationArguments args) throws Exception {
|
||||||
|
List<String> list = Collections.singletonList(topic);
|
||||||
|
consumer.subscribe(list);
|
||||||
|
|
||||||
|
while (true){
|
||||||
|
ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofMillis(100));
|
||||||
|
consumerRecords.forEach(record -> {
|
||||||
|
String value = record.value();
|
||||||
|
System.out.println(value);
|
||||||
|
|
||||||
|
JSONObject jsonObject = new JSONObject();
|
||||||
|
jsonObject.put("123","123");
|
||||||
|
|
||||||
|
//事件处理
|
||||||
|
eventPublisher.publishEvent(jsonObject);
|
||||||
|
});
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,19 @@
|
||||||
|
package com.muyu.event.controller;
|
||||||
|
|
||||||
|
import org.springframework.web.bind.annotation.RequestMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.controller
|
||||||
|
* @name：DataController
|
||||||
|
* @date：2024/9/29 20:16
|
||||||
|
*/
|
||||||
|
@RestController
|
||||||
|
@RequestMapping("data")
|
||||||
|
public class DataController {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,53 @@
|
||||||
|
package com.muyu.event.controller;
|
||||||
|
|
||||||
|
|
||||||
|
import com.alibaba.fastjson2.JSONObject;
|
||||||
|
import com.muyu.event.service.TestService;
|
||||||
|
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||||
|
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||||
|
import org.springframework.beans.factory.annotation.Autowired;
|
||||||
|
|
||||||
|
import org.springframework.web.bind.annotation.GetMapping;
|
||||||
|
import org.springframework.web.bind.annotation.RestController;
|
||||||
|
|
||||||
|
import javax.annotation.Resource;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.controller
|
||||||
|
* @name：TestController
|
||||||
|
* @date：2024/9/29 20:58
|
||||||
|
*/
|
||||||
|
@RestController("test")
|
||||||
|
public class TestController {
|
||||||
|
|
||||||
|
@Resource
|
||||||
|
private KafkaProducer kafkaProducer;
|
||||||
|
|
||||||
|
|
||||||
|
private static final String topic="vehicle";
|
||||||
|
|
||||||
|
@GetMapping("send")
|
||||||
|
public String sendKafka(){
|
||||||
|
|
||||||
|
String message="发送一条信息";
|
||||||
|
ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic,message);
|
||||||
|
kafkaProducer.send(producerRecord);
|
||||||
|
|
||||||
|
return "success";
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -1,28 +0,0 @@
|
||||||
package com.muyu.event.domian;
|
|
||||||
|
|
||||||
import lombok.AllArgsConstructor;
|
|
||||||
import lombok.Data;
|
|
||||||
import lombok.NoArgsConstructor;
|
|
||||||
import lombok.experimental.SuperBuilder;
|
|
||||||
|
|
||||||
import java.util.List;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* 事件驱动对象
|
|
||||||
* @Author 刘武
|
|
||||||
* @Data 2024/9/29
|
|
||||||
*/
|
|
||||||
@Data
|
|
||||||
@SuperBuilder
|
|
||||||
@AllArgsConstructor
|
|
||||||
@NoArgsConstructor
|
|
||||||
public class EventActuate {
|
|
||||||
/**
|
|
||||||
* json数据
|
|
||||||
*/
|
|
||||||
private String jsonData;
|
|
||||||
/**
|
|
||||||
* 事件驱动key集合
|
|
||||||
*/
|
|
||||||
private List<String> eventKeys;
|
|
||||||
}
|
|
||||||
|
|
@ -1,25 +0,0 @@
|
||||||
package com.muyu.event.eventDispose;
|
|
||||||
|
|
||||||
|
|
||||||
import com.muyu.event.basics.StartEvent;
|
|
||||||
import com.muyu.event.domian.EventActuate;
|
|
||||||
import org.springframework.context.ApplicationListener;
|
|
||||||
import org.springframework.stereotype.Component;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @author 刘武
|
|
||||||
* @package：
|
|
||||||
* @name：AutoStartupEventListener
|
|
||||||
* @date：2024/9/29
|
|
||||||
*/
|
|
||||||
@Component
|
|
||||||
public class AutoStartupEventListener implements ApplicationListener<StartEvent> {
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void onApplicationEvent(StartEvent event) {
|
|
||||||
|
|
||||||
EventActuate eventActuate = event.getEventActuate();
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
@ -1,37 +0,0 @@
|
||||||
package com.muyu.event.eventDispose;
|
|
||||||
|
|
||||||
|
|
||||||
import com.muyu.event.basics.EventProcessBasics;
|
|
||||||
import lombok.AllArgsConstructor;
|
|
||||||
import lombok.Data;
|
|
||||||
import lombok.EqualsAndHashCode;
|
|
||||||
import lombok.extern.log4j.Log4j2;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @author 刘武
|
|
||||||
* @package：
|
|
||||||
* @name：StorageEvent
|
|
||||||
* @date：2024/9/29
|
|
||||||
*/
|
|
||||||
@EqualsAndHashCode(callSuper = true)
|
|
||||||
@Log4j2
|
|
||||||
@Data
|
|
||||||
@AllArgsConstructor
|
|
||||||
public class StorageEvent extends EventProcessBasics {
|
|
||||||
/**
|
|
||||||
* 事件名称
|
|
||||||
*/
|
|
||||||
private String eventName;
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void handleEvent(String eventKey) {
|
|
||||||
if (eventKey.equals(eventName)){
|
|
||||||
log.info("开始执行 [{}] 事件", eventKey);
|
|
||||||
|
|
||||||
}else if (nextEvent != null){
|
|
||||||
nextEvent.handleEvent(eventKey);
|
|
||||||
}else {
|
|
||||||
log.info("处理结束，最后处理的事件为 [{}]", eventKey);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
@ -0,0 +1,37 @@
|
||||||
|
package com.muyu.event.listener;
|
||||||
|
|
||||||
|
import com.alibaba.fastjson2.JSONObject;
|
||||||
|
import com.muyu.event.basic.EventCustom;
|
||||||
|
import com.muyu.event.basic.EventListener;
|
||||||
|
|
||||||
|
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* 添加数据库事件
|
||||||
|
* @program: cloud-server
|
||||||
|
* @author: cuiyongxing
|
||||||
|
* @create: 2024-09-29 17:34
|
||||||
|
**/
|
||||||
|
public class AddDatabaseListener implements EventListener {
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onEvent(EventCustom event) {
|
||||||
|
|
||||||
|
JSONObject jsonObject = event.getData();
|
||||||
|
List<String> keys = new ArrayList<>();
|
||||||
|
List<String> values = new ArrayList<>();
|
||||||
|
|
||||||
|
jsonObject.forEach((key, value) -> {
|
||||||
|
keys.add(key);
|
||||||
|
values.add((String) value);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void onApplicationEvent(EventCustom event) {
|
||||||
|
onEvent(event);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,19 @@
|
||||||
|
package com.muyu.event.service;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.service.impl
|
||||||
|
* @name：DataService
|
||||||
|
* @date：2024/9/29 20:23
|
||||||
|
*/
|
||||||
|
public interface DataService {
|
||||||
|
|
||||||
|
void warnData(String data);
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,10 @@
|
||||||
|
package com.muyu.event.service;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.service
|
||||||
|
* @name：TestService
|
||||||
|
* @date：2024/9/29 20:59
|
||||||
|
*/
|
||||||
|
public interface TestService {
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,26 @@
|
||||||
|
package com.muyu.event.service.impl;
|
||||||
|
|
||||||
|
import com.muyu.event.service.DataService;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.service.impl
|
||||||
|
* @name：DataServiceImpl
|
||||||
|
* @date：2024/9/29 20:24
|
||||||
|
*/
|
||||||
|
@Service
|
||||||
|
public class DataServiceImpl implements DataService {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
@Override
|
||||||
|
public void warnData(String data) {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
@ -0,0 +1,20 @@
|
||||||
|
package com.muyu.event.service.impl;
|
||||||
|
|
||||||
|
import com.muyu.event.service.TestService;
|
||||||
|
import org.springframework.stereotype.Service;
|
||||||
|
|
||||||
|
/**
|
||||||
|
* @author 刘武
|
||||||
|
* @package：com.muyu.event.service.impl
|
||||||
|
* @name：TestServiceImpl
|
||||||
|
* @date：2024/9/29 21:00
|
||||||
|
*/
|
||||||
|
@Service
|
||||||
|
public class TestServiceImpl implements TestService {
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
|
}
|
||||||
|
|
@ -26,6 +26,7 @@
|
||||||
<module>cloud-event</module>
|
<module>cloud-event</module>
|
||||||
</modules>
|
</modules>
|
||||||
|
|
||||||
|
|
||||||
<artifactId>cloud-modules</artifactId>
|
<artifactId>cloud-modules</artifactId>
|
||||||
<packaging>pom</packaging>
|
<packaging>pom</packaging>
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue