mirror of
https://github.com/kirklin/kkmall.git
synced 2024-09-20 06:46:15 +08:00
软件杯-->自行设计并实现一种消息队列【加锁】
This commit is contained in:
Kirk Lin
parent
670a15edbf
commit
45089fa8b1
110
kkmall-mq/pom.xml
Normal file
110
kkmall-mq/pom.xml
Normal file
|
|
@ -0,0 +1,110 @@
|
|||
<?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 https://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-parent</artifactId>
|
||||
<version>2.5.2</version>
|
||||
<relativePath/> <!-- lookup parent from repository -->
|
||||
</parent>
|
||||
<groupId>name.lkk.kkmall</groupId>
|
||||
<artifactId>kkmall-mq</artifactId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
<name>kkmall-mq</name>
|
||||
<description>KK商城-MQ</description>
|
||||
<properties>
|
||||
<java.version>1.8</java.version>
|
||||
<spring-cloud.version>2020.0.3</spring-cloud.version>
|
||||
</properties>
|
||||
<dependencies>
|
||||
<!-- 每个微服务的公共的依赖，bean，工具类等等-->
|
||||
<dependency>
|
||||
<groupId>name.lkk.kkmall</groupId>
|
||||
<artifactId>kkmall-common</artifactId>
|
||||
<version>0.0.1-SNAPSHOT</version>
|
||||
<exclusions>
|
||||
<!-- 排除 Mybatis-Plus -->
|
||||
<exclusion>
|
||||
<groupId>com.baomidou</groupId>
|
||||
<artifactId>mybatis-plus-boot-starter</artifactId>
|
||||
</exclusion>
|
||||
</exclusions>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-web</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-openfeign</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-loadbalancer</artifactId>
|
||||
</dependency>
|
||||
<!-- https://mvnrepository.com/artifact/com.github.ben-manes.caffeine/caffeine -->
|
||||
<dependency>
|
||||
<groupId>com.github.ben-manes.caffeine</groupId>
|
||||
<artifactId>caffeine</artifactId>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-thymeleaf</artifactId>
|
||||
</dependency>
|
||||
<!-- 使用Redis-->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-data-redis</artifactId>
|
||||
</dependency>
|
||||
<!-- Redisson分布式锁-->
|
||||
<dependency>
|
||||
<groupId>org.redisson</groupId>
|
||||
<artifactId>redisson</artifactId>
|
||||
<version>3.12.5</version>
|
||||
</dependency>
|
||||
<!-- Spring-Cache-->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-cache</artifactId>
|
||||
</dependency>
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>org.springframework.boot</groupId>-->
|
||||
<!-- <artifactId>spring-boot-devtools</artifactId>-->
|
||||
<!-- <optional>true</optional>-->
|
||||
<!-- </dependency>-->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-configuration-processor</artifactId>
|
||||
<optional>true</optional>
|
||||
</dependency>
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
<dependencyManagement>
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>org.springframework.cloud</groupId>
|
||||
<artifactId>spring-cloud-dependencies</artifactId>
|
||||
<version>${spring-cloud.version}</version>
|
||||
<type>pom</type>
|
||||
<scope>import</scope>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
</dependencyManagement>
|
||||
|
||||
<build>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
package name.lkk.kkmall.mq;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
|
||||
import org.springframework.cloud.openfeign.EnableFeignClients;
|
||||
|
||||
@SpringBootApplication
|
||||
@EnableDiscoveryClient
|
||||
@EnableFeignClients("name.lkk.kkmall.mq.feign")
|
||||
public class KkmallMqApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(KkmallMqApplication.class, args);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,56 @@
|
|||
package name.lkk.kkmall.mq.config;
|
||||
|
||||
import name.lkk.kkmall.mq.service.ConsumerService;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
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.StringRedisTemplate;
|
||||
import org.springframework.data.redis.listener.PatternTopic;
|
||||
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
|
||||
import org.springframework.data.redis.listener.adapter.MessageListenerAdapter;
|
||||
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
|
||||
/**
|
||||
* @Description redis消息队列的监听者
|
||||
*/
|
||||
@Configuration
|
||||
public class ListenerConfig {
|
||||
|
||||
@Autowired
|
||||
ThreadPoolExecutor threadPoolExecutor;
|
||||
|
||||
@Bean
|
||||
RedisMessageListenerContainer container(RedisConnectionFactory connectionFactory,
|
||||
MessageListenerAdapter listenerAdapter) {
|
||||
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
|
||||
container.setConnectionFactory(connectionFactory);
|
||||
container.addMessageListener(listenerAdapter, new PatternTopic("lkk"));
|
||||
container.setTaskExecutor(threadPoolExecutor);
|
||||
return container;
|
||||
}
|
||||
|
||||
/**
|
||||
* 利用反射来创建监听到的消息之后执行方法
|
||||
*
|
||||
* @param consumerService
|
||||
* @return
|
||||
*/
|
||||
@Bean
|
||||
MessageListenerAdapter listenerAdapter(ConsumerService consumerService) {
|
||||
return new MessageListenerAdapter(consumerService, "receiveMessage");
|
||||
}
|
||||
|
||||
/**
|
||||
* 使用默认的工厂初始化redis操作模板
|
||||
*
|
||||
* @param connectionFactory
|
||||
* @return
|
||||
*/
|
||||
@Bean
|
||||
StringRedisTemplate template(RedisConnectionFactory connectionFactory) {
|
||||
return new StringRedisTemplate(connectionFactory);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,57 @@
|
|||
package name.lkk.kkmall.mq.config;
|
||||
|
||||
import org.springframework.boot.autoconfigure.cache.CacheProperties;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.cache.annotation.EnableCaching;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.data.redis.cache.RedisCacheConfiguration;
|
||||
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.RedisSerializationContext;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/24 3:32 下午
|
||||
* @description:Spring Cache配置
|
||||
*/
|
||||
@Configuration
|
||||
@EnableCaching
|
||||
@EnableConfigurationProperties(CacheProperties.class)
|
||||
public class MallCacheConfig {
|
||||
/**
|
||||
* 配置文件中 TTL设置没用上
|
||||
*
|
||||
* 原来:
|
||||
* @ConfigurationProperties(prefix = "spring.cache")
|
||||
* public class CacheProperties
|
||||
*
|
||||
* 现在要让这个配置文件生效 : @EnableConfigurationProperties(CacheProperties.class)
|
||||
*
|
||||
*/
|
||||
@Bean
|
||||
RedisCacheConfiguration redisCacheConfiguration(CacheProperties cacheProperties){
|
||||
|
||||
RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig();
|
||||
|
||||
// 设置kv的序列化机制
|
||||
config = config.serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(new StringRedisSerializer()));
|
||||
config = config.serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(new GenericJackson2JsonRedisSerializer()));
|
||||
CacheProperties.Redis redisProperties = cacheProperties.getRedis();
|
||||
|
||||
// 设置配置
|
||||
if(redisProperties.getTimeToLive() != null){
|
||||
config = config.entryTtl(redisProperties.getTimeToLive());
|
||||
}
|
||||
if(redisProperties.getKeyPrefix() != null){
|
||||
config = config.prefixCacheNameWith(redisProperties.getKeyPrefix());
|
||||
}
|
||||
if(!redisProperties.isCacheNullValues()){
|
||||
config = config.disableCachingNullValues();
|
||||
}
|
||||
if(!redisProperties.isUseKeyPrefix()){
|
||||
config = config.disableKeyPrefix();
|
||||
}
|
||||
return config;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
package name.lkk.kkmall.mq.config;
|
||||
|
||||
import org.redisson.Redisson;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.redisson.config.Config;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/23 8:43 下午
|
||||
* @description:
|
||||
*/
|
||||
@Configuration
|
||||
public class MallRedissonConfig {
|
||||
|
||||
@Value("${spring.redis.host}")
|
||||
private String redisHost;
|
||||
@Value("${spring.redis.port}")
|
||||
private String redisPort;
|
||||
|
||||
@Bean(destroyMethod = "shutdown")
|
||||
public RedissonClient redisson() {
|
||||
Config config = new Config();
|
||||
config.useSingleServer().setAddress("redis://"+redisHost+":"+redisPort);
|
||||
return Redisson.create(config);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,33 @@
|
|||
package name.lkk.kkmall.mq.config;
|
||||
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.LinkedBlockingDeque;
|
||||
import java.util.concurrent.ThreadPoolExecutor;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
@Configuration
|
||||
public class MyThreadConfig {
|
||||
|
||||
@Bean
|
||||
public ThreadPoolExecutor threadPoolExecutor(ThreadPoolConfigProperties threadPoolConfigProperties){
|
||||
/**
|
||||
* 线程池创建的七个参数
|
||||
* @param corePoolSize 线程池核心线程大小
|
||||
* @param maximumPoolSize 线程最大线程数量
|
||||
* @param keepAliveTime 空闲线程存活时间
|
||||
* @param unit 空闲线程存活时间单位
|
||||
* @param workQueue 工作队列
|
||||
* @param threadFactory 线程工厂
|
||||
* @param handler 拒绝策略
|
||||
* CallerRunsPolicy 直接拒绝
|
||||
* AbortPolicy 直接丢弃任务并抛异常
|
||||
* DiscardPolicy 直接丢弃任务，并且什么都不做
|
||||
* DiscardOldestPolicy 抛弃进入队列最早的任务，然后尝试把这次拒绝的任务放入队列
|
||||
*
|
||||
*/
|
||||
return new ThreadPoolExecutor(threadPoolConfigProperties.getCoreSize(), threadPoolConfigProperties.getMaxSize(), threadPoolConfigProperties.getKeepAliveTime() , TimeUnit.SECONDS, new LinkedBlockingDeque<>(10000), Executors.defaultThreadFactory(),new ThreadPoolExecutor.AbortPolicy());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
package name.lkk.kkmall.mq.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.RedisTemplate;
|
||||
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
|
||||
|
||||
@Configuration
|
||||
public class RedisConfiguration {
|
||||
|
||||
@Bean
|
||||
@SuppressWarnings("all")
|
||||
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) {
|
||||
RedisTemplate<String, Object> template = new RedisTemplate<String, Object>();
|
||||
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;
|
||||
|
||||
}
|
||||
}
|
||||
616
kkmall-mq/src/main/java/name/lkk/kkmall/mq/config/RedisUtil.java
Normal file
616
kkmall-mq/src/main/java/name/lkk/kkmall/mq/config/RedisUtil.java
Normal file
|
|
@ -0,0 +1,616 @@
|
|||
package name.lkk.kkmall.mq.config;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
import org.springframework.util.CollectionUtils;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* @Description redis的工具类
|
||||
*/
|
||||
@Component
|
||||
public class RedisUtil {
|
||||
|
||||
@Autowired
|
||||
private RedisTemplate<String, Object> redisTemplate;
|
||||
|
||||
/**
|
||||
* 根据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((Collection<String>) 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;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 递增
|
||||
*
|
||||
* @param key 键
|
||||
* @param delta 要增加几(大于0)
|
||||
* @return
|
||||
*/
|
||||
public long incr(String key, long delta) {
|
||||
if (delta < 0) {
|
||||
throw new RuntimeException("递增因子必须大于0");
|
||||
}
|
||||
return redisTemplate.opsForValue().increment(key, delta);
|
||||
}
|
||||
|
||||
/**
|
||||
* 递减
|
||||
*
|
||||
* @param key 键
|
||||
* @param delta 要减少几(小于0)
|
||||
* @return
|
||||
*/
|
||||
public long decr(String key, long delta) {
|
||||
if (delta < 0) {
|
||||
throw new RuntimeException("递减因子必须大于0");
|
||||
}
|
||||
return redisTemplate.opsForValue().increment(key, -delta);
|
||||
}
|
||||
|
||||
/**
|
||||
* 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<Object, Object> hmget(String key) {
|
||||
return redisTemplate.opsForHash().entries(key);
|
||||
}
|
||||
|
||||
/**
|
||||
* HashSet
|
||||
*
|
||||
* @param key 键
|
||||
* @param map 对应多个键值
|
||||
* @return true 成功 false 失败
|
||||
*/
|
||||
public boolean hmset(String key, Map<String, Object> map) {
|
||||
try {
|
||||
redisTemplate.opsForHash().putAll(key, map);
|
||||
return true;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 指定缓存失效时间
|
||||
*
|
||||
* @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;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* HashSet 并设置时间
|
||||
*
|
||||
* @param key 键
|
||||
* @param map 对应多个键值
|
||||
* @param time 时间(秒)
|
||||
* @return true成功 false失败
|
||||
*/
|
||||
public boolean hmset(String key, Map<String, Object> 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);
|
||||
}
|
||||
|
||||
/**
|
||||
* hash递增 如果不存在,就会创建一个 并把新增后的值返回
|
||||
*
|
||||
* @param key 键
|
||||
* @param item 项
|
||||
* @param by 要增加几(大于0)
|
||||
* @return
|
||||
*/
|
||||
public double hincr(String key, String item, double by) {
|
||||
return redisTemplate.opsForHash().increment(key, item, by);
|
||||
}
|
||||
|
||||
/**
|
||||
* hash递减
|
||||
*
|
||||
* @param key 键
|
||||
* @param item 项
|
||||
* @param by 要减少记(小于0)
|
||||
* @return
|
||||
*/
|
||||
public double hdecr(String key, String item, double by) {
|
||||
return redisTemplate.opsForHash().increment(key, item, -by);
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据key获取Set中的所有值
|
||||
*
|
||||
* @param key 键
|
||||
* @return
|
||||
*/
|
||||
public Set<Object> sGet(String key) {
|
||||
try {
|
||||
return redisTemplate.opsForSet().members(key);
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据value从一个set中查询,是否存在
|
||||
*
|
||||
* @param key 键
|
||||
* @param value 值
|
||||
* @return true 存在 false不存在
|
||||
*/
|
||||
public boolean sHasKey(String key, Object value) {
|
||||
try {
|
||||
return redisTemplate.opsForSet().isMember(key, value);
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将数据放入set缓存
|
||||
*
|
||||
* @param key 键
|
||||
* @param values 值 可以是多个
|
||||
* @return 成功个数
|
||||
*/
|
||||
public long sSet(String key, Object... values) {
|
||||
try {
|
||||
return redisTemplate.opsForSet().add(key, values);
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将set数据放入缓存
|
||||
*
|
||||
* @param key 键
|
||||
* @param time 时间(秒)
|
||||
* @param values 值 可以是多个
|
||||
* @return 成功个数
|
||||
*/
|
||||
public long sSetAndTime(String key, long time, Object... values) {
|
||||
try {
|
||||
Long count = redisTemplate.opsForSet().add(key, values);
|
||||
if (time > 0) {
|
||||
expire(key, time);
|
||||
}
|
||||
return count;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取set缓存的长度
|
||||
*
|
||||
* @param key 键
|
||||
* @return
|
||||
*/
|
||||
public long sGetSetSize(String key) {
|
||||
try {
|
||||
return redisTemplate.opsForSet().size(key);
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 移除值为value的
|
||||
*
|
||||
* @param key 键
|
||||
* @param values 值 可以是多个
|
||||
* @return 移除的个数
|
||||
*/
|
||||
public long setRemove(String key, Object... values) {
|
||||
try {
|
||||
Long count = redisTemplate.opsForSet().remove(key, values);
|
||||
return count;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取list缓存的内容
|
||||
*
|
||||
* @param key 键
|
||||
* @param start 开始
|
||||
* @param end 结束 0 到 -1代表所有值
|
||||
* @return
|
||||
*/
|
||||
public List<Object> lGet(String key, long start, long end) {
|
||||
try {
|
||||
return redisTemplate.opsForList().range(key, start, end);
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 获取list缓存的长度
|
||||
*
|
||||
* @param key 键
|
||||
* @return
|
||||
*/
|
||||
public long lGetListSize(String key) {
|
||||
try {
|
||||
return redisTemplate.opsForList().size(key);
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 通过索引 获取list中的值
|
||||
*
|
||||
* @param key 键
|
||||
* @param index 索引 index>=0时， 0 表头，1 第二个元素，依次类推；index<0时，-1，表尾，-2倒数第二个元素，依次类推
|
||||
* @return
|
||||
*/
|
||||
public Object lGetIndex(String key, long index) {
|
||||
try {
|
||||
return redisTemplate.opsForList().index(key, index);
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将list放入缓存
|
||||
*
|
||||
* @param key 键
|
||||
* @param value 值
|
||||
* @return
|
||||
*/
|
||||
public boolean lSet(String key, Object value) {
|
||||
try {
|
||||
redisTemplate.opsForList().rightPush(key, value);
|
||||
return true;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将list放入缓存
|
||||
*
|
||||
* @param key 键
|
||||
* @param value 值
|
||||
* @param time 时间(秒)
|
||||
* @return
|
||||
*/
|
||||
public boolean lSet(String key, Object value, long time) {
|
||||
try {
|
||||
redisTemplate.opsForList().rightPush(key, value);
|
||||
if (time > 0) {
|
||||
expire(key, time);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将list放入缓存
|
||||
*
|
||||
* @param key 键
|
||||
* @param value 值
|
||||
* @return
|
||||
*/
|
||||
public boolean lSet(String key, List<Object> value) {
|
||||
try {
|
||||
redisTemplate.opsForList().rightPushAll(key, value);
|
||||
return true;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 将list放入缓存
|
||||
*
|
||||
* @param key 键
|
||||
* @param value 值
|
||||
* @param time 时间(秒)
|
||||
* @return
|
||||
*/
|
||||
public boolean lSet(String key, List<Object> value, long time) {
|
||||
try {
|
||||
redisTemplate.opsForList().rightPushAll(key, value);
|
||||
if (time > 0) {
|
||||
expire(key, time);
|
||||
}
|
||||
return true;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 根据索引修改list中的某条数据
|
||||
*
|
||||
* @param key 键
|
||||
* @param index 索引
|
||||
* @param value 值
|
||||
* @return
|
||||
*/
|
||||
public boolean lUpdateIndex(String key, long index, Object value) {
|
||||
try {
|
||||
redisTemplate.opsForList().set(key, index, value);
|
||||
return true;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 移除N个值为value
|
||||
*
|
||||
* @param key 键
|
||||
* @param count 移除多少个
|
||||
* @param value 值
|
||||
* @return 移除的个数
|
||||
*/
|
||||
public long lRemove(String key, long count, Object value) {
|
||||
try {
|
||||
Long remove = redisTemplate.opsForList().remove(key, count, value);
|
||||
|
||||
return remove;
|
||||
}
|
||||
catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 加锁
|
||||
*
|
||||
* @param key
|
||||
* @param value
|
||||
* @param expire
|
||||
* @return 是否加锁成功
|
||||
*/
|
||||
public boolean lock(String key, Object value, long expire) {
|
||||
if (null == value) {
|
||||
value = new Byte[]{1};
|
||||
}
|
||||
if (redisTemplate.opsForValue().setIfAbsent(key, value)) {
|
||||
expire(key, expire);
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* 解锁
|
||||
*
|
||||
* @param key
|
||||
*/
|
||||
public void delLock(String key) {
|
||||
redisTemplate.delete(key);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,18 @@
|
|||
package name.lkk.kkmall.mq.config;
|
||||
|
||||
import lombok.Data;
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
|
||||
@ConfigurationProperties(prefix = "kkmall.thread")
|
||||
@Component
|
||||
@Data
|
||||
public class ThreadPoolConfigProperties {
|
||||
|
||||
private Integer coreSize;
|
||||
|
||||
private Integer maxSize;
|
||||
|
||||
private Integer keepAliveTime;
|
||||
}
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
package name.lkk.kkmall.mq.controller;
|
||||
|
||||
import name.lkk.common.utils.R;
|
||||
import name.lkk.kkmall.mq.service.ProducerService;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.web.bind.annotation.PathVariable;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/26 5:02 下午
|
||||
* @description:
|
||||
*/
|
||||
@RestController
|
||||
@RequestMapping("/pub")
|
||||
public class PublisherController {
|
||||
@Autowired
|
||||
ProducerService producerService;
|
||||
|
||||
@RequestMapping("{name}")
|
||||
public String sendMessage(@PathVariable("name") String name) {
|
||||
return producerService.publish("lkk", R.ok("接收到的信息："+name));
|
||||
}
|
||||
}
|
||||
9
kkmall-mq/src/main/java/name/lkk/kkmall/mq/dto/Test.java
Normal file
9
kkmall-mq/src/main/java/name/lkk/kkmall/mq/dto/Test.java
Normal file
|
|
@ -0,0 +1,9 @@
|
|||
package name.lkk.kkmall.mq.dto;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/26 5:35 下午
|
||||
* @description:
|
||||
*/
|
||||
public class Test {
|
||||
}
|
||||
|
|
@ -0,0 +1,10 @@
|
|||
package name.lkk.kkmall.mq.service;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/26 4:16 下午
|
||||
* @description:
|
||||
*/
|
||||
public interface ConsumerService {
|
||||
public void receiveMessage(String message) ;
|
||||
}
|
||||
|
|
@ -0,0 +1,12 @@
|
|||
package name.lkk.kkmall.mq.service;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/26 4:16 下午
|
||||
* @description:
|
||||
*/
|
||||
public interface ProducerService {
|
||||
|
||||
public String publish(String channel, Object msgObj);
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
package name.lkk.kkmall.mq.service.impl;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import name.lkk.common.utils.R;
|
||||
import name.lkk.kkmall.mq.service.ConsumerService;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/26 4:17 下午
|
||||
* @description: 消费者
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class ConsumerServiceImpl implements ConsumerService {
|
||||
@Autowired
|
||||
private RedisTemplate redisTemplate;
|
||||
// private static Logger logger = LoggerFactory.getLogger(ConsumerServiceImpl.class);
|
||||
|
||||
@Override
|
||||
public void receiveMessage(String message) {
|
||||
R r = JSON.parseObject(message, R.class);
|
||||
log.info("消息队列==> "+r);
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,43 @@
|
|||
package name.lkk.kkmall.mq.service.impl;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import name.lkk.kkmall.mq.service.ProducerService;
|
||||
import org.redisson.api.RLock;
|
||||
import org.redisson.api.RedissonClient;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
/**
|
||||
* @author: kirklin
|
||||
* @date: 2021/6/26 4:18 下午
|
||||
* @description: 生产者
|
||||
*/
|
||||
@Service
|
||||
@Slf4j
|
||||
public class ProducerServiceImpl implements ProducerService {
|
||||
@Autowired
|
||||
private StringRedisTemplate stringRedisTemplate;
|
||||
|
||||
@Autowired
|
||||
private RedissonClient redissonClient;
|
||||
|
||||
@Override
|
||||
public String publish(String channel, Object msgObj) {
|
||||
RLock lock = redissonClient.getLock("my-lock");
|
||||
lock.lock();
|
||||
try {
|
||||
log.info("加锁成功");
|
||||
stringRedisTemplate.convertAndSend(channel, JSON.toJSONString(msgObj));
|
||||
return "消息发送成功了"+JSON.toJSONString(msgObj);
|
||||
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
return "消息发送失败了"+e;
|
||||
}finally {
|
||||
lock.unlock();
|
||||
log.info("解锁成功");
|
||||
}
|
||||
}
|
||||
}
|
||||
51
kkmall-mq/src/main/resources/application.yml
Normal file
51
kkmall-mq/src/main/resources/application.yml
Normal file
|
|
@ -0,0 +1,51 @@
|
|||
ipAddr: localhost
|
||||
server:
|
||||
port: 40000
|
||||
spring:
|
||||
cloud:
|
||||
nacos:
|
||||
config:
|
||||
server-addr: localhost:8848
|
||||
application:
|
||||
name: kkmall-mq
|
||||
jackson:
|
||||
date-format: yyyy-MM-dd HH:mm:ss
|
||||
time-zone: GMT+8
|
||||
mvc:
|
||||
static-path-pattern: /static/**
|
||||
thymeleaf:
|
||||
cache: true
|
||||
suffix: .html
|
||||
prefix: classpath:/templates/
|
||||
|
||||
redis:
|
||||
host: ${ipAddr}
|
||||
port: 6379
|
||||
password: linkekun
|
||||
pool:
|
||||
max-active: 100
|
||||
max-idle: 10
|
||||
min-idle: 0
|
||||
max-wait: 100000
|
||||
timeout: 0
|
||||
# 设置缓存类型
|
||||
cache:
|
||||
type: redis
|
||||
# 设置存活时间
|
||||
redis:
|
||||
time-to-live: 3600000
|
||||
# 如果指定了前缀就用我们指定的 如果没有就用缓存的名字作为前缀
|
||||
# key-prefix: CACHE_
|
||||
# 是否缓存空值,解决缓存穿透问题
|
||||
cache-null-values: true
|
||||
# cache-names:
|
||||
|
||||
logging:
|
||||
level:
|
||||
name.lkk.kkmall: debug
|
||||
|
||||
kkmall:
|
||||
thread:
|
||||
core-size: 20
|
||||
max-size: 200
|
||||
keep-alive-time: 10
|
||||
Loading…
Reference in a new issue