diff --git a/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java b/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java index 631ab01..9c8a4f6 100644 --- a/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java +++ b/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java @@ -19,8 +19,8 @@ import java.util.stream.Collectors; /** * 操作kafka的工具类 * - * @author 154594742@qq.com - * @date 2021/3/2 14:52 + * @author zengwei + * @date 2021-09-07 */ @Component diff --git a/common/src/main/java/cn/org/gitlink/notification/common/utils/SpringContextUtil.java b/common/src/main/java/cn/org/gitlink/notification/common/utils/SpringContextUtil.java new file mode 100644 index 0000000..740d33b --- /dev/null +++ b/common/src/main/java/cn/org/gitlink/notification/common/utils/SpringContextUtil.java @@ -0,0 +1,43 @@ +package cn.org.gitlink.notification.common.utils; + +import org.springframework.beans.BeansException; +import org.springframework.context.ApplicationContext; +import org.springframework.context.ApplicationContextAware; +import org.springframework.stereotype.Component; + +/** + * spring context 工具类，主要用来获取context里面的bean + * @author zengwei + * @Date 2021-09-09 + * */ +@Component +public class SpringContextUtil implements ApplicationContextAware { + + private static ApplicationContext applicationContext; + + @Override + public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { + SpringContextUtil.applicationContext = applicationContext; + } + + //获取applicationContext + public static ApplicationContext getApplicationContext() { + return applicationContext; + } + + //通过name获取 Bean. + public static Object getBean(String name) { + return getApplicationContext().getBean(name); + } + + //通过class获取Bean. + public static T getBean(Class clazz) { + return getApplicationContext().getBean(clazz); + } + + //通过name,以及Clazz返回指定的Bean + public static T getBean(String name, Class clazz) { + return getApplicationContext().getBean(name, clazz); + } + +} diff --git a/executor/pom.xml b/executor/pom.xml index 4d4e1c9..bc1f7e1 100644 --- a/executor/pom.xml +++ b/executor/pom.xml @@ -28,6 +28,19 @@ 2.3.0 + + cn.org.gitlink.notification + gns-common + 1.0.0 + + + + cn.org.gitlink.notification + gns-model + 1.0.0 + compile + + diff --git a/executor/src/main/java/cn/org/gitlink/notification/executor/ExecutorApplication.java b/executor/src/main/java/cn/org/gitlink/notification/executor/ExecutorApplication.java index 0ac7dc6..b4d23a1 100644 --- a/executor/src/main/java/cn/org/gitlink/notification/executor/ExecutorApplication.java +++ b/executor/src/main/java/cn/org/gitlink/notification/executor/ExecutorApplication.java @@ -2,7 +2,9 @@ package cn.org.gitlink.notification.executor; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.context.annotation.ComponentScan; +@ComponentScan("cn.org.gitlink.notification.*") @SpringBootApplication public class ExecutorApplication { public static void main(String[] args) { diff --git a/executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java b/executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java new file mode 100644 index 0000000..2456acc --- /dev/null +++ b/executor/src/main/java/cn/org/gitlink/notification/executor/core/config/KafkaConsumerConfig.java @@ -0,0 +1,58 @@ +package cn.org.gitlink.notification.executor.core.config; + +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.kafka.annotation.EnableKafka; + +import java.util.HashMap; +import java.util.Map; + +@Configuration +@EnableKafka +public class KafkaConsumerConfig { + + @Value("${kafka.consumer.servers}") + private String servers; + @Value("${kafka.consumer.enable.auto.commit}") + private boolean enableAutoCommit; + @Value("${kafka.consumer.session.timeout:10000}") + private String sessionTimeout; + @Value("${kafka.consumer.auto.commit.interval}") + private String autoCommitInterval; + @Value("${kafka.consumer.group.id}") + private String groupId; + @Value("${kafka.consumer.auto.offset.reset}") + private String autoOffsetReset; + @Value("${kafka.consumer.max.poll.records}") + private String maxPollRecords; + + @Bean("kafkaConsumer") + public KafkaConsumer consumer() { + return new KafkaConsumer<>(consumerConfigs()); + } + + + public Map consumerConfigs() { + Map props = new HashMap<>(); + props.put("bootstrap.servers", servers); + //每个消费者分配独立的组号 + props.put("group.id", groupId); + //如果value合法，则自动提交偏移量 + props.put("enable.auto.commit", enableAutoCommit); + // 每次拉取10条 + props.put("max.poll.records", maxPollRecords); + //设置多久一次更新被消费消息的偏移量 + props.put("auto.commit.interval.ms", autoCommitInterval); + //设置会话响应的时间，超过这个时间kafka可以选择放弃消费或者消费下一条消息 + props.put("session.timeout.ms", sessionTimeout); + //自动重置offset + props.put("auto.offset.reset", autoOffsetReset); + props.put("key.deserializer", + "org.apache.kafka.common.serialization.StringDeserializer"); + props.put("value.deserializer", + "org.apache.kafka.common.serialization.StringDeserializer"); + return props; + } +} diff --git a/executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/NotificationJob.java b/executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/NotificationJob.java new file mode 100644 index 0000000..8177cec --- /dev/null +++ b/executor/src/main/java/cn/org/gitlink/notification/executor/service/jobhandler/NotificationJob.java @@ -0,0 +1,65 @@ +package cn.org.gitlink.notification.executor.service.jobhandler; + +import cn.org.gitlink.notification.common.utils.SpringContextUtil; +import cn.org.gitlink.notification.model.dao.entity.vo.NewSysNotificationVo; +import cn.org.gitlink.notification.model.service.notification.SysNotificationService; +import com.alibaba.fastjson.JSONObject; +import com.xxl.job.core.context.XxlJobHelper; +import com.xxl.job.core.handler.annotation.XxlJob; +import org.apache.kafka.clients.consumer.ConsumerRecord; +import org.apache.kafka.clients.consumer.ConsumerRecords; +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.time.Duration; +import java.util.Collections; + +/** + * 系统消息消费执行器 + * @author zengwei + * @Date 2021-09-09 + */ +@Component +public class NotificationJob { + private static Logger logger = LoggerFactory.getLogger(NotificationJob.class); + private static String topic = "gitlink_notification"; + + @Autowired + private SysNotificationService sysNotificationService; + + /** + * 系统消息处理入口，在xxl-job-admin添加任务时，JobHandler栏填注解内名字 + * @author zengwei + * @Date 2021-09-09 + * */ + @XxlJob("notificationMessageHandler") + public void notificationMessageHandler() { + KafkaConsumer consumer = (KafkaConsumer) SpringContextUtil.getBean("kafkaConsumer"); + consumer.subscribe(Collections.singletonList(topic)); + ConsumerRecords records = consumer.poll(Duration.ofMillis(100)); + XxlJobHelper.log("拉取到{}条记录！", records.count()); + + for (ConsumerRecord record: records) { + NewSysNotificationVo newSysNotificationVo = JSONObject.parseObject(record.value(), NewSysNotificationVo.class); + try { + sysNotificationService.sendNotification(newSysNotificationVo); + XxlJobHelper.log("{} 消费成功!", record.value()); + } catch (Exception e) { + XxlJobHelper.log("{} 消费失败, 原因:{}", record.value(), e.getMessage()); + } + } + + consumer.commitAsync(); + + } + + public void init(){ + logger.info("init notification job"); + } + public void destroy(){ + logger.info("notification job has been destroyed"); + } +} diff --git a/executor/src/main/resources/application.properties b/executor/src/main/resources/application.properties index 4585a00..98f667c 100644 --- a/executor/src/main/resources/application.properties +++ b/executor/src/main/resources/application.properties @@ -1,7 +1,7 @@ server.port=8083 logging.config=classpath:logback.xml ### xxl-job admin address list, such as "http://address" or "http://address01,http://address02" -xxl.job.admin.addresses=http://127.0.0.1:8080/xxl-job-admin +xxl.job.admin.addresses=http://127.0.0.1:9999/xxl-job-admin ### xxl-job, access token xxl.job.accessToken= ### xxl-job executor appname @@ -15,3 +15,16 @@ xxl.job.executor.port=82 xxl.job.executor.logpath=./log/xxl-job/jobhandler ### xxl-job executor log-retention-days xxl.job.executor.logretentiondays=30 + +kafka.consumer.servers=172.16.15.29:9092 +kafka.consumer.group.id=0 +kafka.consumer.auto.offset.reset=earliest +kafka.consumer.enable.auto.commit=false +kafka.consumer.auto.commit.interval=100 +kafka.consumer.max.poll.records =10 + +# Դ +spring.datasource.driver-class-name=com.mysql.jdbc.Driver +spring.datasource.url=jdbc:mysql://127.0.0.1:33306/gitlink_notification?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8&allowMultiQueries=true +spring.datasource.username=gitlink +spring.datasource.password=giTlinK0^827 \ No newline at end of file diff --git a/writer/pom.xml b/writer/pom.xml index 386307e..4c5a87d 100644 --- a/writer/pom.xml +++ b/writer/pom.xml @@ -32,12 +32,6 @@ 1.0.0 compile - - cn.org.gitlink.notification - gns-model - 1.0.0 - compile - diff --git a/writer/src/main/resources/application.yml b/writer/src/main/resources/application.yml index e7cf0a3..609e97d 100644 --- a/writer/src/main/resources/application.yml +++ b/writer/src/main/resources/application.yml @@ -40,7 +40,7 @@ spring: datasource: # 配置数据源类型 driver-class-name: com.mysql.jdbc.Driver - url: jdbc:mysql://localhost:33306/gitlink_notification?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8&allowMultiQueries=true + url: jdbc:mysql://127.0.0.1:33306/gitlink_notification?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&serverTimezone=GMT%2B8&allowMultiQueries=true username: gitlink password: giTlinK0^827 #ip白名单列表，多个ip用逗号隔开，允许所有用*号