3Redis与Java的集成
Posted ＊King＊
tags:
篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了3Redis与Java的集成相关的知识，希望对你有一定的参考价值。
1、Maven配置
<dependency>
<groupId>redis.clients</groupId>
<artifactId>jedis</artifactId>
<version>3.6.3</version>
</dependency>
2、application.properties配置文件
# Redis服务器地址
redis.host=1.15.106.108
# Redis服务器连接端口
redis.port=6379
# Redis服务器连接密码（默认为空）
redis.password=null
redis.timeout=30000
# 连接池最大连接数（使用负值表示没有限制）
redis.maxTotal=30
# 连接池中的最大空闲连接
redis.maxIdle=10
redis.numTestsPerEvictionRun=1024
redis.timeBetweenEvictionRunsMillis=30000
redis.minEvictableIdleTimeMillis=1800000
redis.softMinEvictableIdleTimeMillis=10000
# 连接池最大阻塞等待时间（使用负值表示没有限制）
redis.maxWaitMillis=1500
redis.testOnBorrow=true
redis.testWhileIdle=true
redis.blockWhenExhausted=false
redis.JmxEnabled=true
3、RedisConfig
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.PropertySource;
import redis.clients.jedis.JedisPool;
import redis.clients.jedis.JedisPoolConfig;
@Configuration
@PropertySource("classpath:application.properties")
public class RedisConfig
@Value("$redis.host")
private String host;
@Value("$redis.port")
private int port;
@Value("$redis.timeout")
private int timeout;
@Value("$redis.maxIdle")
private int maxIdle;
@Value("$redis.maxWaitMillis")
private int maxWaitMillis;
@Value("$redis.blockWhenExhausted")
private Boolean blockWhenExhausted;
@Value("$redis.JmxEnabled")
private Boolean jmxEnabled;
@Bean
public JedisPool jedisPoolFactory()
JedisPoolConfig jedisPoolConfig = new JedisPoolConfig();
jedisPoolConfig.setMaxIdle(maxIdle);
jedisPoolConfig.setMaxWaitMillis(maxWaitMillis);
//连接耗尽时是否阻塞，false报异常，true阻塞直到超时，默认为true
jedisPoolConfig.setBlockWhenExhausted(blockWhenExhausted);
//是否启用pool的jmx管理功能，默认为true
jedisPoolConfig.setJmxEnabled(jmxEnabled);
JedisPool jedisPool = new JedisPool(jedisPoolConfig,host,port,timeout);
return jedisPool;
4、List实现消息简单的队列
如果业务需求比较简单，想把redis当作队列来使用，可以使用List这个数据类型来实现，因为List底层的实现就是一个链表，在头部和尾部操作元素，时间复杂度都是O(1)，它非常符合消息队列的模型。如果把List当作队列，生产者使用lpush发布消息，消费者使用rpop拉取消息。在拉取消息时，如果使用一个死循环来不断地从队列中拉取消息进行处理就会造成CPU空转，不仅浪费CPU，还会对Redis造成压力。所以在消息端可以使用redis提供的阻塞读blpop和brpop（b代表blocking），阻塞读在队列没有数据的时候进入休眠状态，一旦数据到来则立刻醒过来，消息延迟几乎为零。
缺点就是做消费者确认ACK麻烦，不能保证消费者消费消息后是否成功处理的问题，通常需要维护一个Pending列表，来保证消息处理确认。不能做广播模式，如消息发布/订阅模型，不能重复消费，一旦消费就会被删除，不支持分组消费。
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.util.List;
@Component
public class ListVer
public final static String RS_LIST_MQ_NS = "rlm:";
@Autowired
private JedisPool jedisPool;
/*消费者接受消息*/
public List<String> get(String key)
Jedis jedis = null;
try
jedis = jedisPool.getResource();
return jedis.brpop(0,RS_LIST_MQ_NS +key);
catch (Exception e)
throw new RuntimeException("接受消息失败！");
finally
jedis.close();
/*生产者发送消息*/
public void put(String key, String message)
Jedis jedis = null;
try
jedis = jedisPool.getResource();
jedis.lpush(RS_LIST_MQ_NS+key,message);
catch (Exception e)
throw new RuntimeException("发送消息失败！");
finally
jedis.close();
5、测试类
import cn.enjoyedu.redis.redismq.ListVer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import java.util.List;
@SpringBootTest
public class TestListVer
@Autowired
private ListVer listVer;
@Test
void testGet()
List<String> result = listVer.get("listmq");
for(String message : result)
System.out.println(message);
@Test
void testPut()
listVer.put("listmq","msgtest");
6、测试结果
7、有序集合ZSET实现消息队列
ZSET更多的是用来实现延时队列，将消息序列化成一个字符串作为zset的value，这个消息的到期处理时间作为score，然后用多个线程轮询zset获取到期的任务进行处理。
Redis 的 zrem 方法是多线程多进程争抢任务的关键，它的返回值决定了当 前实例有没有抢到任务，因为 获取消息的方法可能会被多个线程、多个进程调 用，同一个任务可能会被多个进程线程抢到，迪过 zrem 来决定唯一的属主。
缺点就是消费者必须使用轮询的方式。
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@Component
public class ZSetVer
public final static String RS_ZS_MQ_NS = "rzsm:";
@Autowired
private JedisPool jedisPool;
/*生产者，消息的发送，实际生产中，相关参数，
比如订单信息，过期时间等应该传入，可以考虑将订单信息json化存入redis*/
public void producer()
Jedis jedis = null;
try
jedis = jedisPool.getResource();
for (int i = 0; i < 5; i++)
String order_id = "000000000"+i;
double score = System.currentTimeMillis()+(i*1000);
jedis.zadd(RS_ZS_MQ_NS+"orderId",score, order_id);
System.out.println("生产订单: " + order_id + " 当前时间："
+ new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()));
System.out.println((3 + i) + "秒后执行");
catch (Exception e)
throw new RuntimeException("生产消息失败！");
finally
jedis.close();
//消费者，取订单
public void consumerDelayMessage()
Jedis jedis = null;
try
jedis = jedisPool.getResource();
while (true)
Set<String> order = jedis.zrangeByScore(RS_ZS_MQ_NS+"orderId", 0,
System.currentTimeMillis(), 0,1);
if (order == null || order.isEmpty())
System.out.println("当前没有等待的任务");
try
TimeUnit.MICROSECONDS.sleep(1000);
catch (InterruptedException e)
e.printStackTrace();
continue;
String s = order.iterator().next();
if (jedis.zrem(RS_ZS_MQ_NS+"orderId", s)>0)
/*业务处理*/
System.out.println(s);
catch (Exception e)
throw new RuntimeException("消费消息失败！");
finally
jedis.close();
8、测试类
import cn.enjoyedu.redis.redismq.ZSetVer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
public class TestZSetVer
@Autowired
private ZSetVer zSetVer;
@Test
void testConsumerDelayMessage()
zSetVer.consumerDelayMessage();
@Test
void testProducer()
zSetVer.producer();
9、测试结果
以上是关于3Redis与Java的集成的主要内容，如果未能解决你的问题，请参考以下文章