From 73c16974bef683c8f25a1eb1571743f4b614b7e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=9B=BE=E4=BC=9F?= Date: Tue, 7 Sep 2021 10:17:38 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E5=85=AC=E5=85=B1=E7=BB=84?= =?UTF-8?q?=E4=BB=B6=201.=20=E5=B0=81=E8=A3=85redis=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E7=B1=BB=202.=20=E5=B0=81=E8=A3=85kafka=E5=B7=A5=E5=85=B7?= =?UTF-8?q?=E7=B1=BB?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- common/pom.xml | 62 +++++ .../common/config/RedisConfig.java | 44 ++++ .../notification/common/utils/KafkaUtil.java | 104 ++++++++ .../notification/common/utils/RedisUtil.java | 236 ++++++++++++++++++ pom.xml | 1 + reader/pom.xml | 6 +- .../reader/ReaderApiApplication.java | 2 + .../reader/controller/IndexController.java | 7 +- reader/src/main/resources/application.yml | 24 +- writer/pom.xml | 5 + .../notification/WriterApiApplication.java | 2 + writer/src/main/resources/application.yml | 35 ++- 12 files changed, 524 insertions(+), 4 deletions(-) create mode 100644 common/pom.xml create mode 100644 common/src/main/java/cn/org/gitlink/notification/common/config/RedisConfig.java create mode 100644 common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java create mode 100644 common/src/main/java/cn/org/gitlink/notification/common/utils/RedisUtil.java diff --git a/common/pom.xml b/common/pom.xml new file mode 100644 index 0000000..b43adc5 --- /dev/null +++ b/common/pom.xml @@ -0,0 +1,62 @@ + + + + gitlink-notification-system + cn.org.gitlink.notification + 1.0.0 + + gns-common + 1.0.0 + + + + org.apache.maven.plugins + maven-compiler-plugin + + 8 + 8 + + + + + [GNS] - Common + jar + api + 4.0.0 + + + + + org.springframework.boot + spring-boot-starter-data-redis + 2.3.4.RELEASE + + + com.alibaba + fastjson + 1.2.3 + + + + com.fasterxml.jackson.core + jackson-core + 2.11.2 + + + + com.fasterxml.jackson.core + jackson-databind + 2.11.2 + + + + + org.springframework.kafka + spring-kafka + 2.7.6 + + + + \ No newline at end of file diff --git a/common/src/main/java/cn/org/gitlink/notification/common/config/RedisConfig.java b/common/src/main/java/cn/org/gitlink/notification/common/config/RedisConfig.java new file mode 100644 index 0000000..ed9adf4 --- /dev/null +++ b/common/src/main/java/cn/org/gitlink/notification/common/config/RedisConfig.java @@ -0,0 +1,44 @@ +package cn.org.gitlink.notification.common.config; + +import com.fasterxml.jackson.annotation.JsonAutoDetect; +import com.fasterxml.jackson.annotation.PropertyAccessor; +import com.fasterxml.jackson.databind.ObjectMapper; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.RedisConnectionFactory; +import org.springframework.data.redis.core.*; +import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer; +import org.springframework.data.redis.serializer.StringRedisSerializer; + +/** + * redis配置类 + * @Date: 2021/09/03 + * @Author: zengwei + * @Description: + */ +@Configuration +public class RedisConfig { + + @Bean + @SuppressWarnings("all") + public RedisTemplate redisTemplate(RedisConnectionFactory factory) { + RedisTemplate template = new RedisTemplate(); + template.setConnectionFactory(factory); + Jackson2JsonRedisSerializer jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer(Object.class); + ObjectMapper om = new ObjectMapper(); + om.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY); + om.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL); + jackson2JsonRedisSerializer.setObjectMapper(om); + StringRedisSerializer stringRedisSerializer = new StringRedisSerializer(); + // key采用String的序列化方式 + template.setKeySerializer(stringRedisSerializer); + // hash的key也采用String的序列化方式 + template.setHashKeySerializer(stringRedisSerializer); + // value序列化方式采用jackson + template.setValueSerializer(jackson2JsonRedisSerializer); + // hash的value序列化方式采用jackson + template.setHashValueSerializer(jackson2JsonRedisSerializer); + template.afterPropertiesSet(); + return template; + } +} 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 new file mode 100644 index 0000000..631ab01 --- /dev/null +++ b/common/src/main/java/cn/org/gitlink/notification/common/utils/KafkaUtil.java @@ -0,0 +1,104 @@ +package cn.org.gitlink.notification.common.utils; +import com.google.common.collect.Lists; +import org.apache.kafka.clients.admin.*; +import org.apache.kafka.common.TopicPartitionInfo; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.kafka.core.KafkaTemplate; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.util.Collection; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; + +/** + * 操作kafka的工具类 + * + * @author 154594742@qq.com + * @date 2021/3/2 14:52 + */ + +@Component +public class KafkaUtil { + + @Value("${spring.kafka.bootstrap-servers:#{null}}") + private String springKafkaBootstrapServers; + + private AdminClient adminClient; + + @Autowired + private KafkaTemplate kafkaTemplate; + + + /** + * 初始化AdminClient + * '@PostConstruct该注解被用来修饰一个非静态的void（）方法。 + * 被@PostConstruct修饰的方法会在服务器加载Servlet的时候运行，并且只会被服务器执行一次。 + * PostConstruct在构造函数之后执行，init（）方法之前执行。 + */ + @PostConstruct + private void initAdminClient() { + Map props = new HashMap<>(1); + if (springKafkaBootstrapServers == null) { + return; + } + props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, springKafkaBootstrapServers); + adminClient = KafkaAdminClient.create(props); + } + + /** + * 新增topic，支持批量 + */ + public void createTopic(Collection newTopics) { + adminClient.createTopics(newTopics); + } + + /** + * 删除topic，支持批量 + */ + public void deleteTopic(Collection topics) { + adminClient.deleteTopics(topics); + } + + /** + * 获取指定topic的信息 + */ + public String getTopicInfo(Collection topics) { + AtomicReference info = new AtomicReference(""); + try { + adminClient.describeTopics(topics).all().get().forEach((topic, description) -> { + for (TopicPartitionInfo partition : description.partitions()) { + info.set(info + partition.toString() + "\n"); + } + }); + } catch (InterruptedException | ExecutionException e) { + e.printStackTrace(); + } + return info.get(); + } + + /** + * 获取全部topic + */ + public List getAllTopic() { + try { + return adminClient.listTopics().listings().get().stream().map(TopicListing::name).collect(Collectors.toList()); + } catch (InterruptedException | ExecutionException e) { + e.printStackTrace(); + } + return Lists.newArrayList(); + } + + /** + * 往topic中发送消息 + */ + public void sendMessage(String topic, String message) { + kafkaTemplate.send(topic, message); + } + +} \ No newline at end of file diff --git a/common/src/main/java/cn/org/gitlink/notification/common/utils/RedisUtil.java b/common/src/main/java/cn/org/gitlink/notification/common/utils/RedisUtil.java new file mode 100644 index 0000000..487de15 --- /dev/null +++ b/common/src/main/java/cn/org/gitlink/notification/common/utils/RedisUtil.java @@ -0,0 +1,236 @@ +package cn.org.gitlink.notification.common.utils; + +import java.util.Map; +import java.util.concurrent.TimeUnit; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.data.redis.core.RedisTemplate; +import org.springframework.stereotype.Component; +import org.springframework.util.CollectionUtils; +/** + * Redis工具类 + * @author zengwei + * @date 2021/09/03 + */ +@Component +public class RedisUtil { + + @Autowired + private RedisTemplate redisTemplate; + + // =============================common============================ + /** + * 指定缓存失效时间 + * @param key 键 + * @param time 时间(秒) + * @return + */ + public boolean expire(String key, long time) { + try { + if (time > 0) { + redisTemplate.expire(key, time, TimeUnit.SECONDS); + } + return true; + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 根据key 获取过期时间 + * @param key 键 不能为null + * @return 时间(秒) 返回0代表为永久有效 + */ + public long getExpire(String key) { + return redisTemplate.getExpire(key, TimeUnit.SECONDS); + } + + /** + * 判断key是否存在 + * @param key 键 + * @return true 存在 false不存在 + */ + public boolean hasKey(String key) { + try { + return redisTemplate.hasKey(key); + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 删除缓存 + * @param key 可以传一个值 或多个 + */ + @SuppressWarnings("unchecked") + public void del(String... key) { + if (key != null && key.length > 0) { + if (key.length == 1) { + redisTemplate.delete(key[0]); + } else { + redisTemplate.delete(CollectionUtils.arrayToList(key)); + } + } + } + + // ============================String============================= + /** + * 普通缓存获取 + * @param key 键 + * @return 值 + */ + public Object get(String key) { + return key == null ? null : redisTemplate.opsForValue().get(key); + } + + /** + * 普通缓存放入 + * @param key 键 + * @param value 值 + * @return true成功 false失败 + */ + public boolean set(String key, Object value) { + try { + redisTemplate.opsForValue().set(key, value); + return true; + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 普通缓存放入并设置时间 + * @param key 键 + * @param value 值 + * @param time 时间(秒) time要大于0 如果time小于等于0 将设置无限期 + * @return true成功 false 失败 + */ + public boolean set(String key, Object value, long time) { + try { + if (time > 0) { + redisTemplate.opsForValue().set(key, value, time, TimeUnit.SECONDS); + } else { + set(key, value); + } + return true; + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + // ================================Map================================= + /** + * HashGet + * @param key 键 不能为null + * @param item 项 不能为null + * @return 值 + */ + public Object hget(String key, String item) { + return redisTemplate.opsForHash().get(key, item); + } + + /** + * 获取hashKey对应的所有键值 + * @param key 键 + * @return 对应的多个键值 + */ + public Map hmget(String key) { + return redisTemplate.opsForHash().entries(key); + } + + /** + * HashSet + * @param key 键 + * @param map 对应多个键值 + * @return true 成功 false 失败 + */ + public boolean hmset(String key, Map map) { + try { + redisTemplate.opsForHash().putAll(key, map); + return true; + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * HashSet 并设置时间 + * @param key 键 + * @param map 对应多个键值 + * @param time 时间(秒) + * @return true成功 false失败 + */ + public boolean hmset(String key, Map map, long time) { + try { + redisTemplate.opsForHash().putAll(key, map); + if (time > 0) { + expire(key, time); + } + return true; + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 向一张hash表中放入数据,如果不存在将创建 + * @param key 键 + * @param item 项 + * @param value 值 + * @return true 成功 false失败 + */ + public boolean hset(String key, String item, Object value) { + try { + redisTemplate.opsForHash().put(key, item, value); + return true; + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 向一张hash表中放入数据,如果不存在将创建 + * @param key 键 + * @param item 项 + * @param value 值 + * @param time 时间(秒) 注意:如果已存在的hash表有时间,这里将会替换原有的时间 + * @return true 成功 false失败 + */ + public boolean hset(String key, String item, Object value, long time) { + try { + redisTemplate.opsForHash().put(key, item, value); + if (time > 0) { + expire(key, time); + } + return true; + } catch (Exception e) { + e.printStackTrace(); + return false; + } + } + + /** + * 删除hash表中的值 + * @param key 键 不能为null + * @param item 项 可以使多个 不能为null + */ + public void hdel(String key, Object... item) { + redisTemplate.opsForHash().delete(key, item); + } + + /** + * 判断hash表中是否有该项的值 + * @param key 键 不能为null + * @param item 项 不能为null + * @return true 存在 false不存在 + */ + public boolean hHasKey(String key, String item) { + return redisTemplate.opsForHash().hasKey(key, item); + } +} \ No newline at end of file diff --git a/pom.xml b/pom.xml index 44f1bf3..d89cbf9 100644 --- a/pom.xml +++ b/pom.xml @@ -14,6 +14,7 @@ writer reader executor + common diff --git a/reader/pom.xml b/reader/pom.xml index 8eab4be..9c70c9f 100644 --- a/reader/pom.xml +++ b/reader/pom.xml @@ -20,6 +20,11 @@ spring-boot-starter-web ${springboot.version} + + cn.org.gitlink.notification + gns-common + 1.0.0 + @@ -34,5 +39,4 @@ - \ No newline at end of file diff --git a/reader/src/main/java/cn/org/gitlink/notification/reader/ReaderApiApplication.java b/reader/src/main/java/cn/org/gitlink/notification/reader/ReaderApiApplication.java index f599154..62e69c7 100644 --- a/reader/src/main/java/cn/org/gitlink/notification/reader/ReaderApiApplication.java +++ b/reader/src/main/java/cn/org/gitlink/notification/reader/ReaderApiApplication.java @@ -2,7 +2,9 @@ package cn.org.gitlink.notification.reader; 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 ReaderApiApplication { public static void main(String[] args) { diff --git a/reader/src/main/java/cn/org/gitlink/notification/reader/controller/IndexController.java b/reader/src/main/java/cn/org/gitlink/notification/reader/controller/IndexController.java index 49840a3..78461b4 100644 --- a/reader/src/main/java/cn/org/gitlink/notification/reader/controller/IndexController.java +++ b/reader/src/main/java/cn/org/gitlink/notification/reader/controller/IndexController.java @@ -1,12 +1,17 @@ package cn.org.gitlink.notification.reader.controller; +import cn.org.gitlink.notification.common.utils.RedisUtil; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RestController; @RestController public class IndexController { + @Autowired + private RedisUtil redisUtil; @GetMapping(value = "") public String index() { - return "reader index!"; + redisUtil.set("hello", "Hello reader!"); + return redisUtil.get("hello").toString(); } } diff --git a/reader/src/main/resources/application.yml b/reader/src/main/resources/application.yml index 78fcdc0..95e323d 100644 --- a/reader/src/main/resources/application.yml +++ b/reader/src/main/resources/application.yml @@ -1,2 +1,24 @@ server: - port: 8080 \ No newline at end of file + port: 8080 + +spring: + redis: + # Redis数据库索引（默认为0） + database: 0 + # Redis服务器地址 + host: 127.0.0.1 + # Redis服务器连接端口 + port: 6379 + # Redis服务器连接密码（默认为空） + password: + pool: + # 连接池最大连接数（使用负值表示没有限制） + max-active: 200 + # 连接池最大阻塞等待时间（使用负值表示没有限制） + max-wait: -1 + # 连接池中的最大空闲连接 + max-idle: 10 + # 连接池中的最小空闲连接 + min-idle: 0 + # 连接超时时间（毫秒） + timeout: 1000 diff --git a/writer/pom.xml b/writer/pom.xml index 1f39328..c0584e4 100644 --- a/writer/pom.xml +++ b/writer/pom.xml @@ -21,6 +21,11 @@ spring-boot-starter-web ${springboot.version} + + cn.org.gitlink.notification + gns-common + 1.0.0 + diff --git a/writer/src/main/java/cn/org/gitlink/notification/WriterApiApplication.java b/writer/src/main/java/cn/org/gitlink/notification/WriterApiApplication.java index 8ca037b..fe6aa1f 100644 --- a/writer/src/main/java/cn/org/gitlink/notification/WriterApiApplication.java +++ b/writer/src/main/java/cn/org/gitlink/notification/WriterApiApplication.java @@ -2,7 +2,9 @@ package cn.org.gitlink.notification; 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 WriterApiApplication { public static void main(String[] args) { diff --git a/writer/src/main/resources/application.yml b/writer/src/main/resources/application.yml index c09df25..29d2c7d 100644 --- a/writer/src/main/resources/application.yml +++ b/writer/src/main/resources/application.yml @@ -1,2 +1,35 @@ server: - port: 8082 \ No newline at end of file + port: 8082 + +spring: + #kafka配置 + kafka: + #这里改为你的kafka服务器ip和端口号 + bootstrap-servers: 172.16.15.29:9092 + #=============== producer ======================= + producer: + #如果该值大于零时，表示启用重试失败的发送次数 + retries: 0 + #每当多个记录被发送到同一分区时，生产者将尝试将记录一起批量处理为更少的请求，默认值为16384(单位字节) + batch-size: 16384 + #生产者可用于缓冲等待发送到服务器的记录的内存总字节数，默认值为3355443 + buffer-memory: 33554432 + #key的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializer + key-serializer: org.apache.kafka.common.serialization.StringSerializer + #value的Serializer类，实现类实现了接口org.apache.kafka.common.serialization.Serializer + value-serializer: org.apache.kafka.common.serialization.StringSerializer + #=============== consumer ======================= + consumer: + #用于标识此使用者所属的使用者组的唯一字符串 + group-id: test-consumer-group + #当Kafka中没有初始偏移量或者服务器上不再存在当前偏移量时该怎么办，默认值为latest，表示自动将偏移重置为最新的偏移量 + #可选的值为latest, earliest, none + auto-offset-reset: earliest + #消费者的偏移量将在后台定期提交，默认值为true + enable-auto-commit: true + #如果'enable-auto-commit'为true，则消费者偏移自动提交给Kafka的频率（以毫秒为单位），默认值为5000。 + auto-commit-interval: 100 + #密钥的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializer + key-deserializer: org.apache.kafka.common.serialization.StringDeserializer + #值的反序列化器类，实现类实现了接口org.apache.kafka.common.serialization.Deserializer + value-deserializer: org.apache.kafka.common.serialization.StringDeserializer \ No newline at end of file -- 2.34.1