系统消息executor
1. 添加系统消息executor消息gitlink_notification topic 2. 封装kafka消费者操作 3. DB相关配置修改
This commit is contained in:
曾伟
parent
e690967207
commit
9157e163b0
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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> T getBean(Class<T> clazz) {
|
||||
return getApplicationContext().getBean(clazz);
|
||||
}
|
||||
|
||||
//通过name,以及Clazz返回指定的Bean
|
||||
public static <T> T getBean(String name, Class<T> clazz) {
|
||||
return getApplicationContext().getBean(name, clazz);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -28,6 +28,19 @@
|
|||
<version>2.3.0</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>cn.org.gitlink.notification</groupId>
|
||||
<artifactId>gns-common</artifactId>
|
||||
<version>1.0.0</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>cn.org.gitlink.notification</groupId>
|
||||
<artifactId>gns-model</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
|
|
|
|||
|
|
@ -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) {
|
||||
|
|
|
|||
|
|
@ -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<String, String> consumer() {
|
||||
return new KafkaConsumer<>(consumerConfigs());
|
||||
}
|
||||
|
||||
|
||||
public Map<String, Object> consumerConfigs() {
|
||||
Map<String, Object> 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;
|
||||
}
|
||||
}
|
||||
|
|
@ -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<String, String> consumer = (KafkaConsumer<String, String>) SpringContextUtil.getBean("kafkaConsumer");
|
||||
consumer.subscribe(Collections.singletonList(topic));
|
||||
ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
|
||||
XxlJobHelper.log("拉取到{}条记录！", records.count());
|
||||
|
||||
for (ConsumerRecord<String, String> 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");
|
||||
}
|
||||
}
|
||||
|
|
@ -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
|
||||
|
|
@ -32,12 +32,6 @@
|
|||
<version>1.0.0</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>cn.org.gitlink.notification</groupId>
|
||||
<artifactId>gns-model</artifactId>
|
||||
<version>1.0.0</version>
|
||||
<scope>compile</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
|
|
|
|||
|
|
@ -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用逗号隔开，允许所有用*号
|
||||
|
|
|
|||
Loading…
Reference in New Issue